Meta-Harness
A minimal outer loop that delegates selection AND mutation to a skill-steered proposer over an append-only candidate history, returning a scalar or task-configured Pareto frontier.
"""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/<name>.py`` sources, ``evolution_summary.jsonl``,
``frontier_val.json``, ``reports/<name>.md``, ``logs/<name>/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/<name>.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/<name>.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 ``<run_dir>/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/<name>.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/<name>/{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/<name>.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