diff --git a/evolution_kernel/cli.py b/evolution_kernel/cli.py index 51b01b2..5288faf 100644 --- a/evolution_kernel/cli.py +++ b/evolution_kernel/cli.py @@ -16,6 +16,7 @@ import argparse import json +import subprocess import sys from datetime import datetime, timezone from pathlib import Path @@ -144,7 +145,10 @@ def _run_loop( governor: Governor, goal: dict, ) -> int: - """Run until hard stops trigger. Each iteration saves state immediately.""" + """Run until hard stops trigger or goal is reached.""" + iteration = 0 + pending_strategy: dict | None = None + while True: state = hard_stops.load_state(args.ledger) allowed, why = hard_stops.precheck( @@ -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) + if new_state.halted: _record_halted(args.ledger, new_state, new_state.halt_reason) return 3 @@ -250,5 +266,53 @@ def _print_result(result, *, halted: bool = False, halt_reason: str | None = Non print(json.dumps(payload, indent=2, sort_keys=True)) +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 + + +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 + + if __name__ == "__main__": raise SystemExit(main()) diff --git a/evolution_kernel/config.py b/evolution_kernel/config.py index 03651df..4f08ec5 100644 --- a/evolution_kernel/config.py +++ b/evolution_kernel/config.py @@ -74,6 +74,8 @@ class Roles: planner: tuple[str, ...] = () executor: tuple[str, ...] = () evaluator: tuple[str, ...] = () + goal_evaluator: tuple[str, ...] = () + strategist: tuple[str, ...] = () @dataclass(frozen=True) @@ -93,6 +95,17 @@ class HistoryConfig: max_entries: int = 10 +@dataclass(frozen=True) +class GoalEvaluatorConfig: + enabled: bool = False + + +@dataclass(frozen=True) +class StrategistConfig: + enabled: bool = False + every_n_rounds: int = 3 + + @dataclass(frozen=True) class EvolutionConfig: mission: str @@ -103,6 +116,8 @@ class EvolutionConfig: llm: LLMConfig = field(default_factory=LLMConfig) coding_agent: CodingAgentConfig = field(default_factory=CodingAgentConfig) history: HistoryConfig = field(default_factory=HistoryConfig) + goal_evaluator: GoalEvaluatorConfig = field(default_factory=GoalEvaluatorConfig) + strategist: StrategistConfig = field(default_factory=StrategistConfig) raw: Mapping[str, Any] = field(default_factory=dict) @@ -135,6 +150,8 @@ def parse_config(raw: Mapping[str, Any]) -> EvolutionConfig: llm = _parse_llm(raw.get("llm", {})) coding_agent = _parse_coding_agent(raw.get("coding_agent", {})) history = _parse_history(raw.get("history", {})) + goal_evaluator = _parse_goal_evaluator(raw.get("goal_evaluator", {})) + strategist = _parse_strategist(raw.get("strategist", {})) return EvolutionConfig( mission=mission.strip(), @@ -145,6 +162,8 @@ def parse_config(raw: Mapping[str, Any]) -> EvolutionConfig: llm=llm, coding_agent=coding_agent, history=history, + goal_evaluator=goal_evaluator, + strategist=strategist, raw=dict(raw), ) @@ -212,7 +231,13 @@ def _argv(label: str) -> tuple[str, ...]: f"`roles.{label}` must be a string or a list of non-empty strings" ) - return Roles(planner=_argv("planner"), executor=_argv("executor"), evaluator=_argv("evaluator")) + return Roles( + planner=_argv("planner"), + executor=_argv("executor"), + evaluator=_argv("evaluator"), + goal_evaluator=_argv("goal_evaluator"), + strategist=_argv("strategist"), + ) def _parse_hard_stops(value: Any) -> HardStops: @@ -273,3 +298,18 @@ def _parse_history(value: Any) -> HistoryConfig: if not isinstance(max_entries, int) or isinstance(max_entries, bool) or max_entries < 1: raise ConfigError("`history.max_entries` must be a positive integer") return HistoryConfig(max_entries=max_entries) + + +def _parse_goal_evaluator(value: Any) -> GoalEvaluatorConfig: + if not isinstance(value, Mapping): + raise ConfigError("`goal_evaluator` must be a mapping") + return GoalEvaluatorConfig(enabled=bool(value.get("enabled", False))) + + +def _parse_strategist(value: Any) -> StrategistConfig: + if not isinstance(value, Mapping): + raise ConfigError("`strategist` must be a mapping") + every_n = value.get("every_n_rounds", 3) + if not isinstance(every_n, int) or isinstance(every_n, bool) or every_n < 1: + raise ConfigError("`strategist.every_n_rounds` must be a positive integer") + return StrategistConfig(enabled=bool(value.get("enabled", False)), every_n_rounds=every_n) diff --git a/evolution_kernel/governor.py b/evolution_kernel/governor.py index 1583261..61047b7 100644 --- a/evolution_kernel/governor.py +++ b/evolution_kernel/governor.py @@ -64,7 +64,7 @@ def __init__( self.config_snapshot = dict(config_snapshot) if config_snapshot else None 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: self._ensure_git_repo() self._ensure_accepted_branch() @@ -87,20 +87,20 @@ def run_once(self, goal: Mapping[str, Any], run_id: str | None = None) -> RunRes observation = collect_observation(self.evidence_sources, self.target_repo) write_observation(observation_path, observation) - self._write_json( - run_dir / "planner_input.json", - { - "run_id": run_id, - "goal": goal, - "accepted_branch": ACCEPTED_BRANCH, - "baseline_commit": baseline_commit, - "worktree": str(worktree), - "ledger_dir": str(self.ledger_dir), - "observation_path": str(observation_path), - "allowed_paths": list(self.allowed_paths), - "history": self._build_history(), - }, - ) + planner_input: dict = { + "run_id": run_id, + "goal": goal, + "accepted_branch": ACCEPTED_BRANCH, + "baseline_commit": baseline_commit, + "worktree": str(worktree), + "ledger_dir": str(self.ledger_dir), + "observation_path": str(observation_path), + "allowed_paths": list(self.allowed_paths), + "history": self._build_history(), + } + if strategy is not None: + planner_input["strategy"] = strategy + self._write_json(run_dir / "planner_input.json", planner_input) self._run_role(self.planner, run_dir / "planner_input.json", run_dir / "plan.json", worktree) self._write_json( diff --git a/roles/goal_evaluator.py b/roles/goal_evaluator.py new file mode 100644 index 0000000..3085a1c --- /dev/null +++ b/roles/goal_evaluator.py @@ -0,0 +1,106 @@ +#!/usr/bin/env python3 +"""Goal evaluator role. + +Reads goal_eval_input.json, calls an LLM to decide whether the overall mission +is complete, and writes goal_evaluation.json. + +LLM provider/model are read from config.json in the same run directory (same +pattern as roles/planner.py). +""" +from __future__ import annotations + +import argparse +import json +import os +import re +import sys +from pathlib import Path + + +def _call_anthropic(prompt: str, model: str, api_key_env: str) -> str: + import anthropic # type: ignore + client = anthropic.Anthropic(api_key=os.environ[api_key_env]) + msg = client.messages.create( + model=model, + max_tokens=512, + messages=[{"role": "user", "content": prompt}], + ) + return msg.content[0].text + + +def _call_openai(prompt: str, model: str, api_key_env: str) -> str: + import openai # type: ignore + client = openai.OpenAI(api_key=os.environ[api_key_env]) + resp = client.chat.completions.create( + model=model, + messages=[{"role": "user", "content": prompt}], + ) + return resp.choices[0].message.content + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--input", required=True) + parser.add_argument("--output", required=True) + parser.add_argument("--worktree", required=True) + args = parser.parse_args() + + inp = json.loads(Path(args.input).read_text(encoding="utf-8")) + + run_dir = Path(args.input).parent + cfg = {} + config_path = run_dir / "config.json" + if config_path.exists(): + cfg = json.loads(config_path.read_text(encoding="utf-8")) + llm_cfg = cfg.get("llm", {}) + provider = llm_cfg.get("provider", "anthropic") + model = llm_cfg.get("model", "claude-sonnet-4-6") + api_key_env = llm_cfg.get("api_key_env", "ANTHROPIC_API_KEY") + + mission = inp.get("mission", "") + latest_eval = inp.get("latest_evaluation", {}) + metrics = latest_eval.get("metrics", {}) + + prompt = f"""You are a goal evaluator for an automated code evolution system. + +Mission: {mission} + +Latest evaluation metrics: +{json.dumps(metrics, indent=2)} + +Based on the mission statement and the latest evaluation metrics, has the mission +been fully accomplished? + +Respond with ONLY a JSON object: +- "goal_reached": true if the mission is fully accomplished, false otherwise +- "confidence": a float between 0.0 and 1.0 +- "reason": one sentence explaining your decision +""" + + if provider == "anthropic": + text = _call_anthropic(prompt, model, api_key_env) + elif provider == "openai": + text = _call_openai(prompt, model, api_key_env) + else: + print(f"error: unknown llm.provider: {provider!r}", file=sys.stderr) + sys.exit(1) + + 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 = {"goal_reached": False, "confidence": 0.0, "reason": text[:200]} + + result.setdefault("goal_reached", False) + result.setdefault("confidence", 0.0) + result.setdefault("reason", "") + + Path(args.output).write_text(json.dumps(result, indent=2, sort_keys=True) + "\n", encoding="utf-8") + + +if __name__ == "__main__": + main() diff --git a/roles/strategist.py b/roles/strategist.py new file mode 100644 index 0000000..74d6af9 --- /dev/null +++ b/roles/strategist.py @@ -0,0 +1,105 @@ +#!/usr/bin/env python3 +"""Strategist role. + +Reads strategist_input.json, calls an LLM to produce a high-level strategy +(current stage, next milestone, taboo directions), and writes strategy.json. + +LLM provider/model are read from config.json in the same run directory (same +pattern as roles/planner.py). +""" +from __future__ import annotations + +import argparse +import json +import os +import re +import sys +from pathlib import Path + + +def _call_anthropic(prompt: str, model: str, api_key_env: str) -> str: + import anthropic # type: ignore + client = anthropic.Anthropic(api_key=os.environ[api_key_env]) + msg = client.messages.create( + model=model, + max_tokens=512, + messages=[{"role": "user", "content": prompt}], + ) + return msg.content[0].text + + +def _call_openai(prompt: str, model: str, api_key_env: str) -> str: + import openai # type: ignore + client = openai.OpenAI(api_key=os.environ[api_key_env]) + resp = client.chat.completions.create( + model=model, + messages=[{"role": "user", "content": prompt}], + ) + return resp.choices[0].message.content + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--input", required=True) + parser.add_argument("--output", required=True) + parser.add_argument("--worktree", required=True) + args = parser.parse_args() + + inp = json.loads(Path(args.input).read_text(encoding="utf-8")) + + run_dir = Path(args.input).parent + cfg = {} + config_path = run_dir / "config.json" + if config_path.exists(): + cfg = json.loads(config_path.read_text(encoding="utf-8")) + llm_cfg = cfg.get("llm", {}) + provider = llm_cfg.get("provider", "anthropic") + model = llm_cfg.get("model", "claude-sonnet-4-6") + api_key_env = llm_cfg.get("api_key_env", "ANTHROPIC_API_KEY") + + 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 provider == "anthropic": + text = _call_anthropic(prompt, model, api_key_env) + elif provider == "openai": + text = _call_openai(prompt, model, api_key_env) + else: + print(f"error: unknown llm.provider: {provider!r}", file=sys.stderr) + sys.exit(1) + + 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", []) + + Path(args.output).write_text(json.dumps(result, indent=2, sort_keys=True) + "\n", encoding="utf-8") + + +if __name__ == "__main__": + main() diff --git a/tests/fixtures/goal_evaluator_not_reached.py b/tests/fixtures/goal_evaluator_not_reached.py new file mode 100644 index 0000000..f769546 --- /dev/null +++ b/tests/fixtures/goal_evaluator_not_reached.py @@ -0,0 +1,14 @@ +import argparse +import json +from pathlib import Path + +parser = argparse.ArgumentParser() +parser.add_argument("--input", required=True) +parser.add_argument("--output", required=True) +parser.add_argument("--worktree", required=True) +args = parser.parse_args() + +Path(args.output).write_text( + json.dumps({"goal_reached": False, "confidence": 0.0, "reason": "fixture: never reached"}, indent=2) + "\n", + encoding="utf-8", +) diff --git a/tests/fixtures/goal_evaluator_reached.py b/tests/fixtures/goal_evaluator_reached.py new file mode 100644 index 0000000..60cfb18 --- /dev/null +++ b/tests/fixtures/goal_evaluator_reached.py @@ -0,0 +1,14 @@ +import argparse +import json +from pathlib import Path + +parser = argparse.ArgumentParser() +parser.add_argument("--input", required=True) +parser.add_argument("--output", required=True) +parser.add_argument("--worktree", required=True) +args = parser.parse_args() + +Path(args.output).write_text( + json.dumps({"goal_reached": True, "confidence": 1.0, "reason": "fixture: always reached"}, indent=2) + "\n", + encoding="utf-8", +) diff --git a/tests/fixtures/strategist.py b/tests/fixtures/strategist.py new file mode 100644 index 0000000..daa798a --- /dev/null +++ b/tests/fixtures/strategist.py @@ -0,0 +1,18 @@ +import argparse +import json +from pathlib import Path + +parser = argparse.ArgumentParser() +parser.add_argument("--input", required=True) +parser.add_argument("--output", required=True) +parser.add_argument("--worktree", required=True) +args = parser.parse_args() + +Path(args.output).write_text( + json.dumps({ + "stage": "fixture-stage", + "next_milestone": "fixture milestone", + "taboo_directions": ["do not break tests"], + }, indent=2) + "\n", + encoding="utf-8", +) diff --git a/tests/test_issue10.py b/tests/test_issue10.py new file mode 100644 index 0000000..d5a0fb7 --- /dev/null +++ b/tests/test_issue10.py @@ -0,0 +1,258 @@ +"""Tests for Issue #10: Goal Evaluator + Strategist.""" +from __future__ import annotations + +import json +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +from evolution_kernel import hard_stops +from evolution_kernel.config import parse_config +from evolution_kernel.governor import ACCEPTED_BRANCH, Governor, RoleCommand + + +ROOT = Path(__file__).resolve().parents[1] +FIXTURES = ROOT / "tests" / "fixtures" + + +def _git(args, cwd): + result = subprocess.run(["git", *args], cwd=cwd, text=True, capture_output=True, check=False) + if result.returncode != 0: + raise AssertionError(f"git {' '.join(args)} failed: {result.stderr}") + return result.stdout.strip() + + +def _bootstrap_repo(repo: Path) -> None: + repo.mkdir(parents=True, exist_ok=True) + _git(["init"], repo) + _git(["config", "user.email", "test@example.com"], repo) + _git(["config", "user.name", "Test"], repo) + (repo / "README.md").write_text("# target\n") + src = repo / "src" + src.mkdir(exist_ok=True) + (src / ".gitkeep").write_text("") + _git(["add", "-A"], repo) + _git(["commit", "-m", "initial"], repo) + + +def _role(name: str) -> RoleCommand: + return RoleCommand([sys.executable, str(FIXTURES / name)]) + + +# --------------------------------------------------------------------------- +# Config parsing — new fields +# --------------------------------------------------------------------------- + +class TestNewConfigFields(unittest.TestCase): + + def test_goal_evaluator_defaults(self): + cfg = parse_config({"mission": "x"}) + self.assertFalse(cfg.goal_evaluator.enabled) + + def test_goal_evaluator_enabled(self): + cfg = parse_config({"mission": "x", "goal_evaluator": {"enabled": True}}) + self.assertTrue(cfg.goal_evaluator.enabled) + + def test_strategist_defaults(self): + cfg = parse_config({"mission": "x"}) + self.assertFalse(cfg.strategist.enabled) + self.assertEqual(cfg.strategist.every_n_rounds, 3) + + def test_strategist_custom(self): + cfg = parse_config({"mission": "x", "strategist": {"enabled": True, "every_n_rounds": 5}}) + self.assertTrue(cfg.strategist.enabled) + self.assertEqual(cfg.strategist.every_n_rounds, 5) + + def test_strategist_every_n_rounds_invalid(self): + from evolution_kernel.config import ConfigError + with self.assertRaises(ConfigError): + parse_config({"mission": "x", "strategist": {"every_n_rounds": 0}}) + + def test_roles_goal_evaluator_parsed(self): + cfg = parse_config({"mission": "x", "roles": {"goal_evaluator": ["python3", "eval.py"], + "planner": ["p"], "executor": ["e"], "evaluator": ["ev"]}}) + self.assertEqual(cfg.roles.goal_evaluator, ("python3", "eval.py")) + + def test_roles_strategist_parsed(self): + cfg = parse_config({"mission": "x", "roles": {"strategist": ["python3", "strat.py"], + "planner": ["p"], "executor": ["e"], "evaluator": ["ev"]}}) + self.assertEqual(cfg.roles.strategist, ("python3", "strat.py")) + + +# --------------------------------------------------------------------------- +# Governor — strategy injection +# --------------------------------------------------------------------------- + +class TestStrategyInjection(unittest.TestCase): + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.base = Path(self._tmp.name) + self.repo = self.base / "repo" + self.ledger = self.base / "ledger" + _bootstrap_repo(self.repo) + + def tearDown(self): + self._tmp.cleanup() + + def _make_governor(self) -> Governor: + return Governor( + target_repo=self.repo, + ledger_dir=self.ledger, + planner=_role("planner.py"), + executor=_role("executor.py"), + evaluator=_role("evaluator_accept.py"), + ) + + def test_strategy_appears_in_planner_input(self): + gov = self._make_governor() + strategy = {"stage": "test", "next_milestone": "m1", "taboo_directions": []} + gov.run_once({"name": "t", "objective": "t"}, strategy=strategy) + planner_input = json.loads( + (self.ledger / "runs" / "0001" / "planner_input.json").read_text() + ) + self.assertEqual(planner_input["strategy"], strategy) + + def test_no_strategy_key_when_none(self): + gov = self._make_governor() + gov.run_once({"name": "t", "objective": "t"}) + planner_input = json.loads( + (self.ledger / "runs" / "0001" / "planner_input.json").read_text() + ) + self.assertNotIn("strategy", planner_input) + + +# --------------------------------------------------------------------------- +# CLI — goal_reached exit path +# --------------------------------------------------------------------------- + +class TestGoalReached(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, 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} +""") + 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) + + +if __name__ == "__main__": + unittest.main()