feat: goal evaluator + strategist (closes #10) - #13
Conversation
Two new optional roles that let an evolution session stop on success instead of running to max_iterations: - goal_evaluator: after each accepted run, asks an LLM whether the mission is complete; on goal_reached=true the CLI exits 0 with halted=false. Wired in config under `goal_evaluator.enabled` and `roles.goal_evaluator`. - strategist: every N rounds, produces stage / next_milestone / taboo_directions and injects them into the next planner_input.json. Wired in config under `strategist.enabled`, `strategist.every_n_rounds`, and `roles.strategist`. Hard stop machinery is untouched; both roles are off by default so existing configs keep behaving the same. 15 new tests; full suite 54/54 passing.
There was a problem hiding this comment.
Pull request overview
Adds two optional LLM-driven roles for the evolution loop: a Goal Evaluator that allows early CLI exit (rc=0, halted=false) when the mission is judged complete, and a Strategist that injects high-level strategy into the planner every N rounds. Both roles are off by default, so existing behaviour is preserved.
Changes:
- New
GoalEvaluatorConfig/StrategistConfig+Roles.goal_evaluator/Roles.strategistargv tuples inconfig.py, with validation forevery_n_rounds. Governor.run_oncenow accepts an optionalstrategyand conditionally adds it toplanner_input.json; CLI_run_looporchestrates goal-evaluator check after accepted runs and strategist invocation every N iterations.- New LLM role scripts
roles/goal_evaluator.pyandroles/strategist.py(planner.py-style), plus 3 deterministic test fixtures and 15 new unit tests intests/test_issue10.py.
Reviewed changes
Copilot reviewed 9 out of 9 changed files in this pull request and generated 8 comments.
Show a summary per file
| File | Description |
|---|---|
evolution_kernel/config.py |
Adds GoalEvaluatorConfig/StrategistConfig dataclasses, role argv slots, and parsers with validation. |
evolution_kernel/governor.py |
run_once gains optional strategy arg merged into planner_input.json. |
evolution_kernel/cli.py |
_run_loop adds goal-reached early exit and strategist scheduling; new _check_goal_reached / _invoke_strategist helpers. |
roles/goal_evaluator.py |
New LLM role; outputs {goal_reached, confidence, reason} JSON. |
roles/strategist.py |
New LLM role; outputs {stage, next_milestone, taboo_directions} JSON. |
tests/fixtures/goal_evaluator_reached.py |
Deterministic fixture returning goal_reached: true. |
tests/fixtures/goal_evaluator_not_reached.py |
Deterministic fixture returning goal_reached: false. |
tests/fixtures/strategist.py |
Deterministic fixture emitting a fixed strategy payload. |
tests/test_issue10.py |
15 new tests covering config parsing, strategy injection, goal-reached exit path, and strategist cadence. |
Comments suppressed due to low confidence (3)
evolution_kernel/cli.py:314
- Both
_check_goal_reachedand_invoke_strategistswallow subprocess failures silently: a non-zero exit code, missing output file, or malformed JSON causes the helper to returnFalse/Nonewith no message and no recorded artifact. When a user enables these roles and they fail (e.g. missing dependency, API auth error, invalid JSON), the loop will simply behave as if the role was disabled, making the problem very hard to diagnose. Consider at minimum loggingcompleted.stderr(and the failure reason) to stderr, similar to how other failure paths surface diagnostics.
completed = subprocess.run(argv, text=True, capture_output=True, check=False)
if completed.returncode != 0 or not output_path.exists():
return False
try:
data = json.loads(output_path.read_text(encoding="utf-8"))
return bool(data.get("goal_reached", False))
except (json.JSONDecodeError, OSError):
return False
def _invoke_strategist(cfg: EvolutionConfig, result, iteration: int) -> dict | None:
input_path = result.run_dir / "strategist_input.json"
output_path = result.run_dir / "strategy.json"
input_data = {
"mission": cfg.mission,
"current_round": iteration,
"latest_evaluation": dict(result.evaluation),
}
input_path.write_text(json.dumps(input_data, indent=2, sort_keys=True) + "\n", encoding="utf-8")
argv = [
*cfg.roles.strategist,
"--input", str(input_path),
"--output", str(output_path),
"--worktree", str(result.run_dir),
]
completed = subprocess.run(argv, text=True, capture_output=True, check=False)
if completed.returncode != 0 or not output_path.exists():
return None
try:
return json.loads(output_path.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError):
return None
evolution_kernel/cli.py:290
_check_goal_reachedwritesgoal_eval_input.jsonandgoal_evaluation.jsonintoresult.run_dir, but the governor finalizes the run directory contents insiderun_onceand returns. These post-hoc files are not visible to the planner'shistoryview (built from previous run artifacts), and there's no record in the ledger summary that a goal evaluation was performed. For auditability, consider either having the governor own the goal-evaluator invocation (so it is part of the canonical run artifacts and history), or writing a separate marker that downstream tooling can inspect.
def _check_goal_reached(cfg: EvolutionConfig, result) -> bool:
input_path = result.run_dir / "goal_eval_input.json"
output_path = result.run_dir / "goal_evaluation.json"
input_data = {
"mission": cfg.mission,
"latest_evaluation": dict(result.evaluation),
}
input_path.write_text(json.dumps(input_data, indent=2, sort_keys=True) + "\n", encoding="utf-8")
argv = [
*cfg.roles.goal_evaluator,
"--input", str(input_path),
"--output", str(output_path),
"--worktree", str(result.run_dir),
]
completed = subprocess.run(argv, text=True, capture_output=True, check=False)
if completed.returncode != 0 or not output_path.exists():
return False
try:
data = json.loads(output_path.read_text(encoding="utf-8"))
return bool(data.get("goal_reached", False))
except (json.JSONDecodeError, OSError):
return False
roles/strategist.py:99
- If the LLM responds with a JSON snippet whose regex match parses to a non-dict (e.g. a list, number, or string),
resultwill be set to that value and the subsequentresult.setdefault(...)calls will raiseAttributeError, leading to an unhandled exception from the role. Afterjson.loads(...)succeeds, guard withisinstance(result, dict)before callingsetdefault, and fall back to the default-shaped dict otherwise. The same pattern applies toroles/goal_evaluator.pylines 88–100.
m = re.search(r"\{.*\}", text, re.DOTALL)
result = None
if m:
try:
result = json.loads(m.group())
except json.JSONDecodeError:
pass
if result is None:
result = {"stage": "unknown", "next_milestone": text[:200], "taboo_directions": []}
result.setdefault("stage", "unknown")
result.setdefault("next_milestone", "")
result.setdefault("taboo_directions", [])
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| def _write_config(self, goal_evaluator_fixture: str, goal_evaluator_enabled: bool = True) -> Path: | ||
| config_path = self.base / "evolution.yml" | ||
| ge_role = f'["python3", "{FIXTURES}/{goal_evaluator_fixture}"]' | ||
| ge_enabled = "true" if goal_evaluator_enabled else "false" | ||
| config_path.write_text(f""" | ||
| mission: "test goal" | ||
| hard_stops: | ||
| max_iterations: 3 | ||
| max_consecutive_failures: 5 | ||
| roles: | ||
| planner: ["python3", "{FIXTURES}/planner.py"] | ||
| executor: ["python3", "{FIXTURES}/executor.py"] | ||
| evaluator: ["python3", "{FIXTURES}/evaluator_accept.py"] | ||
| goal_evaluator: {ge_role} | ||
| goal_evaluator: | ||
| enabled: {ge_enabled} | ||
| """) | ||
| return config_path | ||
|
|
||
| def _run_cli(self, config_path: Path, *extra_args): | ||
| from evolution_kernel.cli import main | ||
| return main([ | ||
| "--config", str(config_path), | ||
| "--repo", str(self.repo), | ||
| "--ledger", self.ledger, | ||
| *extra_args, | ||
| ]) | ||
|
|
||
| def test_goal_reached_exits_zero(self): | ||
| cfg_path = self._write_config("goal_evaluator_reached.py") | ||
| rc = self._run_cli(cfg_path, "--loop") | ||
| self.assertEqual(rc, 0) | ||
|
|
||
| def test_goal_reached_stops_after_first_accepted(self): | ||
| cfg_path = self._write_config("goal_evaluator_reached.py") | ||
| self._run_cli(cfg_path, "--loop") | ||
| runs = list((Path(self.ledger) / "runs").iterdir()) | ||
| self.assertEqual(len(runs), 1) | ||
|
|
||
| def test_goal_not_reached_continues_to_hard_stop(self): | ||
| cfg_path = self._write_config("goal_evaluator_not_reached.py") | ||
| rc = self._run_cli(cfg_path, "--loop") | ||
| self.assertEqual(rc, 3) | ||
| runs = list((Path(self.ledger) / "runs").iterdir()) | ||
| self.assertEqual(len(runs), 3) | ||
|
|
||
| def test_goal_evaluator_disabled_does_not_stop_early(self): | ||
| cfg_path = self._write_config("goal_evaluator_reached.py", goal_evaluator_enabled=False) | ||
| rc = self._run_cli(cfg_path, "--loop") | ||
| self.assertEqual(rc, 3) | ||
|
|
||
|
|
||
| # --------------------------------------------------------------------------- | ||
| # CLI — strategist injection | ||
| # --------------------------------------------------------------------------- | ||
|
|
||
| class TestStrategistInjection(unittest.TestCase): | ||
|
|
||
| def setUp(self): | ||
| self._tmp = tempfile.TemporaryDirectory() | ||
| self.base = Path(self._tmp.name) | ||
| self.repo = self.base / "repo" | ||
| self.ledger = str(self.base / "ledger") | ||
| _bootstrap_repo(self.repo) | ||
|
|
||
| def tearDown(self): | ||
| self._tmp.cleanup() | ||
|
|
||
| def _write_config(self, every_n: int = 2) -> Path: | ||
| config_path = self.base / "evolution.yml" | ||
| config_path.write_text(f""" | ||
| mission: "test strategist" | ||
| hard_stops: | ||
| max_iterations: 4 | ||
| max_consecutive_failures: 5 | ||
| roles: | ||
| planner: ["python3", "{FIXTURES}/planner.py"] | ||
| executor: ["python3", "{FIXTURES}/executor.py"] | ||
| evaluator: ["python3", "{FIXTURES}/evaluator_accept.py"] | ||
| strategist: ["python3", "{FIXTURES}/strategist.py"] | ||
| strategist: | ||
| enabled: true | ||
| every_n_rounds: {every_n} | ||
| """) |
| @@ -159,7 +163,10 @@ def _run_loop( | |||
| print(json.dumps({"halted": True, "reason": why}, indent=2, sort_keys=True)) | |||
| return 3 | |||
|
|
|||
| result = governor.run_once(goal) | |||
| result = governor.run_once(goal, strategy=pending_strategy) | |||
| pending_strategy = None | |||
| iteration += 1 | |||
|
|
|||
| cost_usd, tokens_used = _safe_cost(result.evaluation) | |||
| new_state = hard_stops.record_outcome( | |||
| state, | |||
| @@ -173,6 +180,15 @@ def _run_loop( | |||
| ) | |||
| hard_stops.save_state(args.ledger, new_state) | |||
| _print_result(result, halted=new_state.halted, halt_reason=new_state.halt_reason) | |||
|
|
|||
| if result.decision.accepted and cfg.goal_evaluator.enabled and cfg.roles.goal_evaluator: | |||
| if _check_goal_reached(cfg, result): | |||
| print(json.dumps({"goal_reached": True, "halted": False}, indent=2, sort_keys=True)) | |||
| return 0 | |||
|
|
|||
| if cfg.strategist.enabled and cfg.roles.strategist and iteration % cfg.strategist.every_n_rounds == 0: | |||
| pending_strategy = _invoke_strategist(cfg, result, iteration) | |||
| self.history_max_entries = history_max_entries | ||
|
|
||
| def run_once(self, goal: Mapping[str, Any], run_id: str | None = None) -> RunResult: | ||
| def run_once(self, goal: Mapping[str, Any], run_id: str | None = None, strategy: dict | None = None) -> RunResult: |
| model=model, | ||
| messages=[{"role": "user", "content": prompt}], | ||
| ) | ||
| return resp.choices[0].message.content |
| mission = inp.get("mission", "") | ||
| current_round = inp.get("current_round", 0) | ||
| latest_eval = inp.get("latest_evaluation", {}) | ||
| metrics = latest_eval.get("metrics", {}) | ||
|
|
||
| prompt = f"""You are a strategist for an automated code evolution system. | ||
|
|
||
| Mission: {mission} | ||
| Current round: {current_round} | ||
| Latest metrics: {json.dumps(metrics)} | ||
|
|
||
| Assess the current evolution stage and produce a strategy for the next phase. | ||
|
|
||
| Respond with ONLY a JSON object: | ||
| - "stage": name of the current evolution stage (e.g. "exploration", "refinement", "convergence") | ||
| - "next_milestone": one concrete measurable milestone to reach next | ||
| - "taboo_directions": list of approaches that have failed or should be avoided | ||
| """ |
| if result.decision.accepted and cfg.goal_evaluator.enabled and cfg.roles.goal_evaluator: | ||
| if _check_goal_reached(cfg, result): | ||
| print(json.dumps({"goal_reached": True, "halted": False}, indent=2, sort_keys=True)) | ||
| return 0 |
| if result.decision.accepted and cfg.goal_evaluator.enabled and cfg.roles.goal_evaluator: | ||
| if _check_goal_reached(cfg, result): | ||
| print(json.dumps({"goal_reached": True, "halted": False}, indent=2, sort_keys=True)) | ||
| return 0 | ||
|
|
||
| if cfg.strategist.enabled and cfg.roles.strategist and iteration % cfg.strategist.every_n_rounds == 0: | ||
| pending_strategy = _invoke_strategist(cfg, result, iteration) |
| def test_goal_reached_exits_zero(self): | ||
| cfg_path = self._write_config("goal_evaluator_reached.py") | ||
| rc = self._run_cli(cfg_path, "--loop") | ||
| self.assertEqual(rc, 0) | ||
|
|
||
| def test_goal_reached_stops_after_first_accepted(self): | ||
| cfg_path = self._write_config("goal_evaluator_reached.py") | ||
| self._run_cli(cfg_path, "--loop") | ||
| runs = list((Path(self.ledger) / "runs").iterdir()) | ||
| self.assertEqual(len(runs), 1) | ||
|
|
||
| def test_goal_not_reached_continues_to_hard_stop(self): | ||
| cfg_path = self._write_config("goal_evaluator_not_reached.py") | ||
| rc = self._run_cli(cfg_path, "--loop") | ||
| self.assertEqual(rc, 3) | ||
| runs = list((Path(self.ledger) / "runs").iterdir()) | ||
| self.assertEqual(len(runs), 3) | ||
|
|
||
| def test_goal_evaluator_disabled_does_not_stop_early(self): | ||
| cfg_path = self._write_config("goal_evaluator_reached.py", goal_evaluator_enabled=False) | ||
| rc = self._run_cli(cfg_path, "--loop") | ||
| self.assertEqual(rc, 3) | ||
|
|
||
|
|
||
| # --------------------------------------------------------------------------- | ||
| # CLI — strategist injection | ||
| # --------------------------------------------------------------------------- | ||
|
|
||
| class TestStrategistInjection(unittest.TestCase): | ||
|
|
||
| def setUp(self): | ||
| self._tmp = tempfile.TemporaryDirectory() | ||
| self.base = Path(self._tmp.name) | ||
| self.repo = self.base / "repo" | ||
| self.ledger = str(self.base / "ledger") | ||
| _bootstrap_repo(self.repo) | ||
|
|
||
| def tearDown(self): | ||
| self._tmp.cleanup() | ||
|
|
||
| def _write_config(self, every_n: int = 2) -> Path: | ||
| config_path = self.base / "evolution.yml" | ||
| config_path.write_text(f""" | ||
| mission: "test strategist" | ||
| hard_stops: | ||
| max_iterations: 4 | ||
| max_consecutive_failures: 5 | ||
| roles: | ||
| planner: ["python3", "{FIXTURES}/planner.py"] | ||
| executor: ["python3", "{FIXTURES}/executor.py"] | ||
| evaluator: ["python3", "{FIXTURES}/evaluator_accept.py"] | ||
| strategist: ["python3", "{FIXTURES}/strategist.py"] | ||
| strategist: | ||
| enabled: true | ||
| every_n_rounds: {every_n} | ||
| """) | ||
| return config_path | ||
|
|
||
| def _run_cli(self, config_path: Path, *extra_args): | ||
| from evolution_kernel.cli import main | ||
| return main([ | ||
| "--config", str(config_path), | ||
| "--repo", str(self.repo), | ||
| "--ledger", self.ledger, | ||
| *extra_args, | ||
| ]) | ||
|
|
||
| def test_strategy_injected_at_round_n_plus_one(self): | ||
| cfg_path = self._write_config(every_n=2) | ||
| self._run_cli(cfg_path, "--loop") | ||
| # Strategist runs after round 2 → strategy appears in round 3's planner_input | ||
| planner_input_3 = json.loads( | ||
| (Path(self.ledger) / "runs" / "0003" / "planner_input.json").read_text() | ||
| ) | ||
| self.assertIn("strategy", planner_input_3) | ||
| self.assertEqual(planner_input_3["strategy"]["stage"], "fixture-stage") | ||
|
|
||
| def test_no_strategy_in_round_one(self): | ||
| cfg_path = self._write_config(every_n=2) | ||
| self._run_cli(cfg_path, "--loop") | ||
| planner_input_1 = json.loads( | ||
| (Path(self.ledger) / "runs" / "0001" / "planner_input.json").read_text() | ||
| ) | ||
| self.assertNotIn("strategy", planner_input_1) |
E2E verification (Scheme B) — both new LLM prompts PASSRan the two new role prompts through real Claude Sonnet 4.6 via `claude -p` (bypasses ANTHROPIC_API_KEY and aider, validates the one thing unit tests can't: real-model output parses with our regex+json.loads). Test inputs: mission = "Improve test coverage of math_utils from 60% to 95%", metrics = {coverage: 0.62, tests_passing: 18/30}. Goal Evaluator```json Strategist```json Observation worth keepingThe model returns JSON wrapped in ```json … ``` fenced blocks. The role scripts' regex `r"{.*}"` with `re.DOTALL` strips the fences correctly — fallback path was not triggered for either prompt. Coverage of the verification matrix
Ready to merge. |
Summary
Implements Issue #10 / Phase 2 — two optional roles that let an evolution session stop on success instead of running to max_iterations:
goal_reached=truethe CLI exits 0 withhalted=false, distinct from the hard-stop exit 3.stage/next_milestone/taboo_directionsand injects them into the next round'splanner_input.json.Hard-stop machinery (
max_iterations/max_total_usd/max_total_tokens) is untouched. Both roles are off by default, so existing configs keep behaving the same.Changes
evolution_kernel/config.pyGoalEvaluatorConfig,StrategistConfig;Rolesgainsgoal_evaluatorandstrategistargv tuplesevolution_kernel/governor.pyrun_once(..., strategy=...)merges strategy intoplanner_input.jsonevolution_kernel/cli.py_run_loopchecksgoal_reachedafter each accepted run, calls strategist every N rounds; new_check_goal_reachedand_invoke_strategisthelpersroles/goal_evaluator.pyroles/planner.py)roles/strategist.pytests/test_issue10.pytests/fixtures/*Config
Test plan
ANTHROPIC_API_KEY) to confirm prompts produce parseable JSON🤖 Generated with Claude Code