"""Meta-Harness Coding-Agent Proposer — the reference-faithful proposer: a coding agent that BROWSES the on-disk archive ``D`` with terminal tools (grep/cat) and writes new candidate files. One module per component (see scaffold.py). This is the ``proposer.mode: "coding_agent"`` path — the faithful port of the reference ``meta_harness.py`` ``propose_claude`` flow, generalized so the agent is a config choice (``proposer.coding_agent``: ``claude_code`` (default) / ``codex`` / ``cursor_agent`` — see coding_agents.py): * The evolving TARGET differs from the reference (galapagos evolves a task's EVOLVE-BLOCK program, not a MemorySystem/AgentHarness wrapping a frozen base model) — this is the one accepted, unavoidable difference. The evolving METHOD is ported as-is: * Every proposal round, the full archive ``D`` is materialized to disk from the in-memory Population + Memory (``candidates/.py`` sources, ``evolution_summary.jsonl``, ``frontier_val.json``, ``reports/.md``, ``logs//trace.txt``, ``current_best.py``, ``TASK.md``), and a headless coding-agent session is launched at ``cwd=D`` steered by the domain SKILL.md (the selected adapter injects it — a real ``--append-system-prompt`` for Claude Code, or prepended to the prompt for agents without such a flag). The agent reads whatever it wants from ``D`` with its Read/Grep/Bash-equivalent tools, then writes ``candidates/.py`` files and a ``pending_eval.json`` manifest — exactly the reference contract, agent-agnostic. * The outer loop then evaluates every valid candidate with the task's own Evaluator; the agent never evaluates. The FIFO is drained before the proposal-round counter advances, so a batch shares one iteration and one archive/parent snapshot exactly as in the reference. Auth/billing posture is per-adapter (Claude Code = subscription-only, API keys stripped — the reference's exact ``os.environ.pop("ANTHROPIC_API_KEY")`` posture; Codex/Cursor use their own auth). The ``llm`` fallback (``proposer.mode: "llm"``, proposer.py) needs no agent CLI and uses an ordinary API model over a serialized view of ``D`` instead. """ from __future__ import annotations import json import logging import re import tempfile import time import uuid from pathlib import Path from ...components.proposer import Env, Proposer from ...records import Genome from .coding_agents import get_adapter from .population import display_name from .proposer import _STALE_KEYS from .validation import validate_candidate_source log = logging.getLogger(__name__) _EFFORT_LEVELS = ("low", "medium", "high", "xhigh", "max") # union accepted by the adapters _NAME_CLEAN = re.compile(r"[^A-Za-z0-9_]+") _DEFAULT_PROPOSE_TIMEOUT_S = 2400 # the reference proposer timeout per round def _safe_name(raw: str, used: set[str]) -> str: """Filesystem-safe, unique candidate filename stem from a display name.""" stem = _NAME_CLEAN.sub("_", raw).strip("_").lower() or "candidate" name, i = stem, 1 while name in used: i += 1 name = f"{stem}_{i}" used.add(name) return name class MetaHarnessCodingAgentProposer(Proposer): """A headless coding agent over D. FIFO-dispenses siblings inside one proposal round. The launch is delegated to a :class:`~.coding_agents.CodingAgentAdapter` selected by ``proposer.coding_agent`` (default ``claude_code``). The reference candidate contract (write ``candidates/.py`` + a ``pending_eval.json`` manifest, DON'T self-evaluate) is stated in the filesystem skill and task prompt. This proposer reads the manifest and post-eval reports the agent leaves behind.""" trajectory_origin = "agent" def __init__(self, candidates_per_proposal: int = 1, coding_agent: str = "claude_code", model: str | None = None, effort: str | None = None, max_turns: int | None = None, wall_timeout_seconds: int | None = None): self.candidates_per_proposal = max(1, int(candidates_per_proposal)) # steering-only self.adapter = get_adapter(coding_agent) # raises on unknown agent (fail fast) self.model = model self.effort = effort self.max_turns = max_turns self.wall_timeout_seconds = wall_timeout_seconds self._queue: list[dict] = [] self._last_proposal_selection: dict | None = None self._last_proposal_event_causes: list[str] = [] self._last_agent_exit_code: int | None = None self.candidate_validator = None self._active_round = 0 self._d_dir: Path | None = None self._tmp_dir: Path | None = None # created only when there is no run_dir (tests) if self.adapter.experimental: log.warning("meta_harness proposer.coding_agent=%r is EXPERIMENTAL (not run-tested); " "CLI flags may need adjustment for your version", coding_agent) # effort is validated against the adapter union here; the adapter validates its own model if self.effort is not None and self.effort not in _EFFORT_LEVELS: raise ValueError( f"meta_harness: invalid proposer.reasoning_effort {self.effort!r}; use one of " f"{', '.join(_EFFORT_LEVELS)} (or null). The adapter maps it to its own scale.") def has_pending_candidates(self) -> bool: """Whether the current coding-agent manifest still has a valid sibling to evaluate.""" return bool(self._queue) # ---- the on-disk archive D (materialized fresh each round from in-memory state) -------------- def d_dir(self, ctx) -> Path: """The archive root. Prefers the path the scaffold published on the signal bus (``blackboard["meta_harness"]["d_dir"]``) so the PromptBuilder and this proposer resolve the SAME dir; falls back to ``/meta_harness_D`` or a cached temp dir (direct use/tests).""" if self._d_dir is not None: return self._d_dir sig = (ctx.blackboard.get("meta_harness", {}) if ctx is not None else {}) or {} published = sig.get("d_dir") if published: self._d_dir = Path(published) else: run_dir = getattr(ctx, "run_dir", None) if ctx is not None else None if run_dir: self._d_dir = Path(run_dir) / "meta_harness_D" else: self._tmp_dir = Path(tempfile.mkdtemp(prefix="galapagos-metaharness-D-")) self._d_dir = self._tmp_dir self._d_dir.mkdir(parents=True, exist_ok=True) return self._d_dir def _materialize_D(self, d_dir: Path, env: Env) -> None: """(Re)write the full archive D to disk from the in-memory Population + Memory so the CLI agent can browse the complete history with grep/cat (the reference's filesystem D).""" sel = env.selection members = list(getattr(sel, "pool", None) or []) frontier = list(getattr(sel, "inspirations", None) or []) parent = getattr(sel, "parent", None) memory = env.memory ctx = env.ctx (d_dir / "candidates").mkdir(exist_ok=True) (d_dir / "reports").mkdir(exist_ok=True) (d_dir / "logs").mkdir(exist_ok=True) # every prior program source → candidates/.py, and a name map for the summary/frontier used: set[str] = set() name_of: dict[str, str] = {} for g in members: fname = _safe_name(display_name(g), used) name_of[g.id] = fname (d_dir / "candidates" / f"{fname}.py").write_text(g.content, encoding="utf-8") # Phase-0 baselines are evidence/frontier inputs, not evolution-summary rows. Proposed # candidates alone populate evolution_summary.jsonl, exactly like the reference. baselines = list(memory.baselines()) if (memory is not None and hasattr(memory, "baselines")) else [] rows = list(memory.rows()) if (memory is not None and hasattr(memory, "rows")) else [] evaluation_records = [*baselines, *rows] with (d_dir / "evolution_summary.jsonl").open("w", encoding="utf-8") as f: for r in rows: f.write(json.dumps({ "iteration": r.get("iteration"), "system": r.get("name"), "combined_score": r.get("score"), "delta": r.get("delta"), "cost": r.get("cost"), "outcome": r.get("outcome"), "axis": r.get("axis"), "hypothesis": r.get("hypothesis"), }, default=str) + "\n") # baseline + proposed-candidate evidence → logs//{result,metrics,artifacts,...} for r in evaluation_records: stem = _NAME_CLEAN.sub("_", str(r.get("name") or "x")).strip("_") or "x" tdir = d_dir / "logs" / stem tdir.mkdir(parents=True, exist_ok=True) metrics = r.get("metrics") if isinstance(r.get("metrics"), dict) else {} artifacts = r.get("artifacts") if isinstance(r.get("artifacts"), dict) else {} per_instance = r.get("per_instance") text_feedback = r.get("text_feedback") result_payload = { "name": r.get("name"), "genome_id": r.get("genome_id"), "iteration": r.get("iteration"), "outcome": r.get("outcome"), "valid": r.get("valid"), "metrics": metrics, "artifacts": artifacts, "per_instance": per_instance, "text_feedback": text_feedback, } (tdir / "result.json").write_text( json.dumps(result_payload, indent=2, ensure_ascii=False, default=str), encoding="utf-8") (tdir / "metrics.json").write_text( json.dumps(metrics, indent=2, ensure_ascii=False, default=str), encoding="utf-8") (tdir / "artifacts.json").write_text( json.dumps(artifacts, indent=2, ensure_ascii=False, default=str), encoding="utf-8") per_instance_path = tdir / "per_instance.json" if per_instance is not None: per_instance_path.write_text( json.dumps(per_instance, indent=2, ensure_ascii=False, default=str), encoding="utf-8") elif per_instance_path.exists(): per_instance_path.unlink() trace = str(r.get("trace") or "") if trace: (tdir / "trace.txt").write_text(trace, encoding="utf-8") # frontier_val.json — scalar best set or Pareto frontier (score-desc). In Pareto mode cost is # looked up from baseline/evolution evidence; ``genome_chars`` is used only when the run # explicitly selected that objective. sig = (ctx.blackboard.get("meta_harness", {}) if ctx is not None else {}) or {} cost_metric = sig.get("cost_metric") def _cost(g): if cost_metric is None: return None for r in evaluation_records: if r.get("name") == display_name(g) and isinstance(r.get("cost"), (int, float)): return r["cost"] return float(len(g.content)) if cost_metric == "genome_chars" else None frontier_rows = [] for genome in frontier: row = {"system": display_name(genome), "combined_score": genome.fitness} cost = _cost(genome) if cost is not None: row["cost"] = cost frontier_rows.append(row) (d_dir / "frontier_val.json").write_text(json.dumps({ "objective_mode": sig.get("objective_mode", "scalar"), "cost_metric": cost_metric, "_pareto": frontier_rows, }, indent=2, default=str), encoding="utf-8") # reports/.md — the <=30-line per-candidate reports reports = list(memory.reports()) if (memory is not None and hasattr(memory, "reports")) else [] for rep in reports: rname = _NAME_CLEAN.sub("_", str(rep.get("name") or "x")).strip("_") or "x" (d_dir / "reports" / f"{rname}.md").write_text(str(rep.get("report") or ""), encoding="utf-8") # current best + task brief if parent is not None: (d_dir / "current_best.py").write_text(parent.content, encoding="utf-8") (d_dir / "TASK.md").write_text(str(getattr(ctx, "task_context", "") or ""), encoding="utf-8") # ---- the coding-agent proposal session (delegated to the selected adapter) ------------------- def _launch(self, d_dir: Path, system_text: str, user_prompt: str) -> int: """Run one headless coding-agent session at ``cwd=d_dir`` via the selected adapter, steered by the SKILL.md. Returns the exit code (124 on timeout). The agent reads D and writes ``candidates/*.py`` + ``pending_eval.json``.""" timeout = int(self.wall_timeout_seconds) if self.wall_timeout_seconds else _DEFAULT_PROPOSE_TIMEOUT_S log.info("meta_harness coding_agent propose — %s (cwd=%s, effort=%s, timeout=%ss)", self.adapter.name, d_dir, self.effort or "agent default", timeout) return self.adapter.launch(d_dir, system_text, user_prompt, model=self.model, effort=self.effort, max_turns=self.max_turns, timeout=timeout, log_name=(f"agent_sessions/round_{self._active_round:04d}/" f"{self.adapter.name}.log")) @staticmethod def _report_snapshot(d_dir: Path) -> dict[str, str]: """Contents before an agent run, used to ignore stale/failed-session report files.""" snapshot: dict[str, str] = {} for path in (d_dir / "reports").glob("*.md"): try: snapshot[path.name] = path.read_text(encoding="utf-8") except OSError: continue return snapshot @staticmethod def _restore_report_snapshot(d_dir: Path, snapshot: dict[str, str]) -> None: """Roll back Step-0 report writes from an unsuccessful agent session.""" report_dir = d_dir / "reports" for path in report_dir.glob("*.md"): if path.name not in snapshot: try: path.unlink() except OSError: pass for name, content in snapshot.items(): try: (report_dir / name).write_text(content, encoding="utf-8") except OSError: pass def _ingest_reports(self, d_dir: Path, memory, before: dict[str, str], *, ctx=None, caused_by_event_id: str | None = None) -> None: """Persist changed ``reports/*.md`` files for candidates that already have evaluation rows. This is the reference skill's Step 0 compression pass. Reports for newly proposed (therefore unevaluated) candidates are deliberately ignored; they become eligible at the next round. """ if memory is None or not hasattr(memory, "rows"): return rows_by_stem: dict[str, dict] = {} for row in memory.rows(): stem = _NAME_CLEAN.sub("_", str(row.get("name") or "x")).strip("_").lower() or "x" rows_by_stem[stem] = row for path in sorted((d_dir / "reports").glob("*.md")): try: report = path.read_text(encoding="utf-8") except OSError: continue if before.get(path.name) == report: continue row = rows_by_stem.get(path.stem.lower()) if row is None: continue report = "\n".join(report.strip().splitlines()[:30]) if report: memory.write("", kind="report", name=str(row.get("name") or path.stem), iteration=int(row.get("iteration") or 0), report=report, replace=True) if ctx is not None: ctx.emit_event( event_type="memory", action="update", status="completed", component_role="memory", iteration=int(row.get("iteration") or 0), evolution_track="solution", caused_by_event_ids=( [caused_by_event_id] if caused_by_event_id else [] ), inputs=([{ "entity_type": "candidate", "entity_id": str(row.get("genome_id")), "role": "report_subject", }] if row.get("genome_id") else []), details={ "memory_name": "candidate_reports", "entry_type": "agent_report", "candidate_name": str(row.get("name") or path.stem), "replace_existing": True, "content": report, }, ) def _collect_candidates(self, d_dir: Path) -> list[dict]: """Read ``pending_eval.json`` + the candidate files the agent wrote (the reference contract). Returns ``[{name, source, report, hypothesis, axis}]``; ``source=None`` marks a manifest entry whose file is missing/unreadable (recorded as a failed row upstream).""" pending = d_dir / "pending_eval.json" if not pending.is_file(): return [] try: data = json.loads(pending.read_text(encoding="utf-8")) except (json.JSONDecodeError, OSError): return [] out: list[dict] = [] for c in (data.get("candidates") or []): if not isinstance(c, dict): continue name = str(c.get("name") or f"candidate_{len(out) + 1}") source = None rel = c.get("file") if rel: fp = (d_dir / str(rel)).resolve() try: # confine reads to D (no path escape) fp.relative_to(d_dir.resolve()) if fp.is_file(): source = fp.read_text(encoding="utf-8") except (ValueError, OSError): source = None if source is None and isinstance(c.get("source"), str): source = c["source"] hyp, axis = str(c.get("hypothesis") or ""), str(c.get("axis") or "") report = "\n".join(x for x in ( f"hypothesis: {hyp}" if hyp else "", f"axis: {axis}" if axis else "", f"changes: {c.get('changes')}" if c.get("changes") else "") if x) out.append({"name": name, "source": source, "report": report, "hypothesis": hyp, "axis": axis}) return out def _refill(self, prompt, env: Env) -> None: """One CLI proposal round: materialize D → launch the agent → collect + compile-gate → queue. Compile-invalid (or file-missing) candidates are recorded as ``failed`` rows immediately, like the chat proposer and the reference's failed candidates.""" d_dir = self.d_dir(env.ctx) iteration = env.ctx.iteration if env.ctx is not None else 0 self._active_round = iteration parent = env.selection.parent proposal_selection = { "inspiration_ids": [ genome.id for genome in env.selection.inspirations if parent is None or genome.id != parent.id ], "pool_size": len(env.selection.pool), } self._last_proposal_selection = proposal_selection # clear a stale manifest from the previous round so we never re-read it try: (d_dir / "pending_eval.json").unlink() except OSError: pass self._materialize_D(d_dir, env) reports_before = self._report_snapshot(d_dir) native_before = { path.resolve() for path in (d_dir / "claude_sessions" / "projects").rglob("*.jsonl") if path.is_file() } agent_started = time.monotonic() agent_event = (env.ctx.emit_event( event_type="agent", action="start", status="completed", component_role="proposer", component_name=type(self).__name__, iteration=iteration, attempt_number=env.attempt_number, evolution_track="solution", caused_by_event_ids=([getattr(env.selection, "_etif_event_id", None)] if getattr(env.selection, "_etif_event_id", None) else []), inputs=([{ "entity_type": "candidate", "entity_id": parent.id, "role": "parent", }] if parent is not None else []), details={ "session_id": f"meta_harness_round_{iteration}", "agent": self.adapter.name, "model": self.model, "round": iteration, }, ) if env.ctx is not None else None) exit_code = self._launch(d_dir, prompt.system or "", prompt.user or "") self._last_agent_exit_code = exit_code round_log = ( d_dir / "agent_sessions" / f"round_{self._active_round:04d}" / f"{self.adapter.name}.log" ) native_after = { path.resolve() for path in (d_dir / "claude_sessions" / "projects").rglob("*.jsonl") if path.is_file() } session_artifacts = ([round_log] if round_log.is_file() else []) + sorted( native_after - native_before ) artifact_root = ( Path(env.ctx.run_dir).resolve() if env.ctx is not None and env.ctx.run_dir else d_dir.resolve() ) session_files: list[str] = [] for path in session_artifacts: try: session_files.append(str(path.resolve().relative_to(artifact_root))) except ValueError: session_files.append(str(path)) agent_finished_event = None if env.ctx is not None: agent_finished_event = env.ctx.emit_event( event_type="agent", action="complete" if exit_code == 0 else "fail", status="completed" if exit_code == 0 else "failed", component_role="proposer", component_name=type(self).__name__, iteration=iteration, attempt_number=env.attempt_number, evolution_track="solution", caused_by_event_ids=[agent_event] if agent_event else [], duration_seconds=time.monotonic() - agent_started, error=(None if exit_code == 0 else { "error_type": "AgentProcessError", "message": f"coding agent exited with status {exit_code}", "traceback": None, }), details={ "session_id": f"meta_harness_round_{iteration}", "agent": self.adapter.name, "model": self.model, "round": iteration, "exit_code": exit_code, "agent_session_format": ( "claude-native" if self.adapter.name == "claude_code" else f"{self.adapter.name}-session-log" ), "native_session_files": session_files, }, ) self._last_proposal_event_causes = [ value for value in (agent_finished_event, agent_event) if value ][:1] if exit_code != 0: # Match the reference propose_claude contract: a failed/timed-out agent round is abandoned # even if it managed to leave a partial manifest behind. Never benchmark partial output. try: (d_dir / "pending_eval.json").unlink() except OSError: pass self._restore_report_snapshot(d_dir, reports_before) log.warning("meta_harness coding-agent proposal failed (exit=%s); discarding round output", exit_code) return self._ingest_reports( d_dir, env.memory, reports_before, ctx=env.ctx, caused_by_event_id=agent_finished_event, ) language = str(getattr(env.ctx, "blackboard", {}).get("language", "python") or "python") for cand in self._collect_candidates(d_dir): source = cand["source"] error = ("no candidate file written (missing/unreadable pending_eval.json entry)" if source is None else validate_candidate_source( source, name=cand["name"], axis=cand.get("axis") or "", hypothesis=cand.get("hypothesis") or "", language=language, validator=self.candidate_validator)) if error: validation_event = None rejected_id = None if env.ctx is not None: rejected_id = f"candidate_{uuid.uuid4().hex}" proposal_event = env.ctx.emit_event( event_type="proposal", action="snapshot", status="completed" if source is not None else "failed", component_role="proposer", component_name=type(self).__name__, iteration=iteration, attempt_number=env.attempt_number, evolution_track="solution", caused_by_event_ids=( [agent_finished_event] if agent_finished_event else [agent_event] if agent_event else [] ), inputs=([{ "entity_type": "candidate", "entity_id": parent.id, "role": "parent", }] if parent is not None else []), details={ "operator": "agent_snapshot", "batch_iteration": iteration, "candidate_name": cand["name"], }, new_candidates=([{ "candidate_id": rejected_id, "candidate_type": "solution", "generation": ( parent.metadata.get("generation", 0) + 1 if parent is not None else 0 ), "parent_ids": [parent.id] if parent is not None else [], "content": source, "attributes": { "candidate_name": cand["name"], "batch_iteration": iteration, }, }] if source is not None else []), ) validation_event = env.ctx.emit_event( event_type="evaluation", action="validate", status="completed" if source is not None else "skipped", component_role="proposer", component_name=type(self).__name__, iteration=iteration, attempt_number=env.attempt_number, evolution_track="solution", caused_by_event_ids=[proposal_event] if proposal_event else [], inputs=([{ "entity_type": "candidate", "entity_id": rejected_id, "role": "candidate", }] if source is not None else []), details={ "evaluation_stage": "interface_validation", "result": {"is_valid": False} if source is not None else None, "reason": error, }, ) if source is not None: env.ctx.emit_event( event_type="population", action="admit", status="skipped", component_role="population", iteration=iteration, attempt_number=env.attempt_number, evolution_track="solution", caused_by_event_ids=( [validation_event] if validation_event else [] ), inputs=[{ "entity_type": "candidate", "entity_id": rejected_id, "role": "candidate", }], details={ "decision": None, "reason": "interface_validation_failed", "new_best": False, }, ) if env.memory is not None: sig = (env.ctx.blackboard.get("meta_harness", {}) if env.ctx is not None else {}) or {} cost = float(len(source or "")) if sig.get("cost_metric") == "genome_chars" else None env.memory.write("", kind="failed", name=cand["name"], iteration=iteration, cost=cost, trace=f"interface validation failed: {error}", axis=cand.get("axis"), hypothesis=cand.get("hypothesis")) if env.ctx is not None: env.ctx.emit_event( event_type="memory", action="write", status="completed", component_role="memory", iteration=iteration, attempt_number=env.attempt_number, evolution_track="solution", caused_by_event_ids=[validation_event] if validation_event else [], inputs=([{ "entity_type": "candidate", "entity_id": rejected_id, "role": "validation_subject", }] if rejected_id and source is not None else []), details={ "memory_name": "evolution_summary", "entry_type": "validation_failure", "candidate_name": cand["name"], "reason": error, }, ) continue cand["_round_parent"] = parent cand["_proposal_selection"] = proposal_selection cand["_proposal_event_causes"] = [ value for value in (agent_finished_event, agent_event) if value ][:1] cand["proposal_round"] = iteration self._queue.append(cand) if self._queue and env.ctx is not None and parent is not None: env.ctx.blackboard.setdefault("meta_harness", {})["proposal_parent_id"] = parent.id # ---- dispense (identical contract to the chat proposer) -------------------------------------- def propose(self, prompt, env: Env) -> Genome: parent = env.selection.parent if not self._queue: self._refill(prompt, env) if not self._queue: # abandoned proposal round (reference: proposer wrote no pending_eval) child = (parent.child(parent.content) if parent is not None else Genome(content="")) for stale in _STALE_KEYS: child.metadata.pop(stale, None) child.metadata["changed"] = False if self._last_proposal_selection is not None: child.trace["proposal_selection"] = self._last_proposal_selection if self._last_proposal_event_causes: child.trace["proposal_caused_by_event_ids"] = list( self._last_proposal_event_causes ) child.trace["etif_no_candidate"] = True child.trace["no_candidate_reason"] = ( "agent_round_failed" if self._last_agent_exit_code else "no_candidates" ) return child cand = self._queue.pop(0) parent = cand.pop("_round_parent", parent) proposal_selection = cand.pop("_proposal_selection", None) proposal_event_causes = cand.pop("_proposal_event_causes", None) if not self._queue and env.ctx is not None: env.ctx.blackboard.setdefault("meta_harness", {}).pop("proposal_parent_id", None) if parent is None: child = Genome(content=cand["source"]) else: child = parent.child(cand["source"], generation=parent.metadata.get("generation", 0) + 1) for stale in _STALE_KEYS: child.metadata.pop(stale, None) child.metadata.update( changed=(parent is None or cand["source"].strip() != parent.content.strip()), candidate_name=cand["name"], proposal_report=cand["report"], axis=cand.get("axis") or "", hypothesis=cand.get("hypothesis") or "", proposal_round=cand.get("proposal_round", env.ctx.iteration if env.ctx else 0), ) if proposal_selection is not None: child.trace["proposal_selection"] = proposal_selection if proposal_event_causes: child.trace["proposal_caused_by_event_ids"] = list(proposal_event_causes) return child