From 89448671824d237d005448f3482dae9729092333 Mon Sep 17 00:00:00 2001 From: Protocol Zero <257158451+Protocol-zero-0@users.noreply.github.com> Date: Wed, 13 May 2026 19:38:29 +0000 Subject: [PATCH] feat: k-branch parallel exploration (closes #14) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Same-round k-branch parallel exploration. Governor.run_once_parallel launches k independent worktrees per round, each running plan → execute → evaluate; the highest-fitness survivor is promoted to evolution/accepted, the rest are recorded under ledger/failed/. - config.parallel.k_branches (default 1); k=1 delegates to run_once - evaluator role emits float fitness; back-compat synthesizes fitness from hard_gates_passed when an older evaluator omits it - per-round cost/tokens are summed across all k branches for hard-stop bookkeeping - 13 new tests (67 total, 54 baseline preserved) covering parity at k=1, k>1 multi-worktree fan-out, fitness ranking, winner promotion, loser ledger demotion, worktree cleanup, cost aggregation, partial scope violation, all-fail no-promotion, and an end-to-end k=3 over 3 rounds CLI loop Co-Authored-By: Claude Opus 4.7 --- README.md | 10 +- README.zh.md | 9 +- evolution_kernel/cli.py | 12 +- evolution_kernel/config.py | 17 ++ evolution_kernel/governor.py | 287 +++++++++++++++++++ roles/evaluator.py | 11 + tests/fixtures/evaluator_branch_fitness.py | 47 ++++ tests/fixtures/evaluator_src_fitness.py | 46 +++ tests/fixtures/executor_branch.py | 31 +++ tests/fixtures/executor_oob_for_run.py | 37 +++ tests/fixtures/executor_unique_marker.py | 24 ++ tests/test_issue14.py | 308 +++++++++++++++++++++ 12 files changed, 833 insertions(+), 6 deletions(-) create mode 100644 tests/fixtures/evaluator_branch_fitness.py create mode 100644 tests/fixtures/evaluator_src_fitness.py create mode 100644 tests/fixtures/executor_branch.py create mode 100644 tests/fixtures/executor_oob_for_run.py create mode 100644 tests/fixtures/executor_unique_marker.py create mode 100644 tests/test_issue14.py diff --git a/README.md b/README.md index 4ee5ca7..8823b56 100644 --- a/README.md +++ b/README.md @@ -245,8 +245,8 @@ flowchart LR | Config-driven: swap LLM provider, model, coding agent | ✅ | | Aider and Claude Code executor support | ✅ | | Anthropic and OpenAI planner/evaluator support | ✅ | -| Goal evaluator — stops when mission is "won" | 🔧 PR #5 | -| k-branch parallel exploration (FunSearch / AlphaEvolve style) | 🔧 PR #6 | +| Goal evaluator — stops when mission is "won" | ✅ | +| k-branch parallel exploration (FunSearch / AlphaEvolve style) | ✅ | | Process sandbox (firejail / bwrap) for production safety | 🔧 PR #7 | --- @@ -293,6 +293,12 @@ coding_agent: history: max_entries: 10 +# Population-level search: per round, spawn k independent worktrees, score +# each branch's fitness, promote the best, demote the rest to ledger/failed/. +# k=1 (default) is plain single-branch run_once behavior. +parallel: + k_branches: 1 + roles: planner: ["python3", "roles/planner.py"] executor: ["bash", "roles/executor.sh"] diff --git a/README.zh.md b/README.zh.md index 21a0959..06bfc98 100644 --- a/README.zh.md +++ b/README.zh.md @@ -245,8 +245,8 @@ flowchart LR | 配置驱动：随时切换 LLM 提供商、模型、coding agent | ✅ | | Aider 和 Claude Code executor 支持 | ✅ | | Anthropic 和 OpenAI 规划器 / 评估器支持 | ✅ | -| 目标评估器——当 mission 完成时自动停止 | 🔧 PR #5 | -| k 路并行探索（FunSearch / AlphaEvolve 模式） | 🔧 PR #6 | +| 目标评估器——当 mission 完成时自动停止 | ✅ | +| k 路并行探索（FunSearch / AlphaEvolve 模式） | ✅ | | 进程级沙箱（firejail / bwrap），面向生产环境 | 🔧 PR #7 | --- @@ -293,6 +293,11 @@ coding_agent: history: max_entries: 10 +# 种群级搜索：每轮起 k 个独立 worktree，按 fitness 排名，最高分推进 evolution/accepted， +# 其余写入 ledger/failed/。k=1（默认）等价于单路 run_once。 +parallel: + k_branches: 1 + roles: planner: ["python3", "roles/planner.py"] executor: ["bash", "roles/executor.sh"] diff --git a/evolution_kernel/cli.py b/evolution_kernel/cli.py index 5288faf..b750b9d 100644 --- a/evolution_kernel/cli.py +++ b/evolution_kernel/cli.py @@ -122,7 +122,11 @@ def _run_with_config(args: argparse.Namespace, cfg: EvolutionConfig) -> int: print(json.dumps({"halted": True, "reason": why}, indent=2, sort_keys=True)) return 3 - result = governor.run_once(goal, run_id=args.run_id) + k = cfg.parallel.k_branches + if k > 1: + result = governor.run_once_parallel(goal, k=k) + else: + result = governor.run_once(goal, run_id=args.run_id) cost_usd, tokens_used = _safe_cost(result.evaluation) new_state = hard_stops.record_outcome( state, @@ -163,7 +167,11 @@ def _run_loop( print(json.dumps({"halted": True, "reason": why}, indent=2, sort_keys=True)) return 3 - result = governor.run_once(goal, strategy=pending_strategy) + k = cfg.parallel.k_branches + if k > 1: + result = governor.run_once_parallel(goal, k=k, strategy=pending_strategy) + else: + result = governor.run_once(goal, strategy=pending_strategy) pending_strategy = None iteration += 1 diff --git a/evolution_kernel/config.py b/evolution_kernel/config.py index 4f08ec5..ff83272 100644 --- a/evolution_kernel/config.py +++ b/evolution_kernel/config.py @@ -106,6 +106,11 @@ class StrategistConfig: every_n_rounds: int = 3 +@dataclass(frozen=True) +class ParallelConfig: + k_branches: int = 1 + + @dataclass(frozen=True) class EvolutionConfig: mission: str @@ -118,6 +123,7 @@ class EvolutionConfig: history: HistoryConfig = field(default_factory=HistoryConfig) goal_evaluator: GoalEvaluatorConfig = field(default_factory=GoalEvaluatorConfig) strategist: StrategistConfig = field(default_factory=StrategistConfig) + parallel: ParallelConfig = field(default_factory=ParallelConfig) raw: Mapping[str, Any] = field(default_factory=dict) @@ -152,6 +158,7 @@ def parse_config(raw: Mapping[str, Any]) -> EvolutionConfig: history = _parse_history(raw.get("history", {})) goal_evaluator = _parse_goal_evaluator(raw.get("goal_evaluator", {})) strategist = _parse_strategist(raw.get("strategist", {})) + parallel = _parse_parallel(raw.get("parallel", {})) return EvolutionConfig( mission=mission.strip(), @@ -164,6 +171,7 @@ def parse_config(raw: Mapping[str, Any]) -> EvolutionConfig: history=history, goal_evaluator=goal_evaluator, strategist=strategist, + parallel=parallel, raw=dict(raw), ) @@ -313,3 +321,12 @@ def _parse_strategist(value: Any) -> StrategistConfig: 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) + + +def _parse_parallel(value: Any) -> ParallelConfig: + if not isinstance(value, Mapping): + raise ConfigError("`parallel` must be a mapping") + k = value.get("k_branches", 1) + if not isinstance(k, int) or isinstance(k, bool) or k < 1: + raise ConfigError("`parallel.k_branches` must be a positive integer") + return ParallelConfig(k_branches=k) diff --git a/evolution_kernel/governor.py b/evolution_kernel/governor.py index 61047b7..ed91a04 100644 --- a/evolution_kernel/governor.py +++ b/evolution_kernel/governor.py @@ -214,6 +214,293 @@ def run_once(self, goal: Mapping[str, Any], run_id: str | None = None, strategy: if worktree.exists(): self._git("worktree", "remove", "--force", str(worktree)) + def run_once_parallel( + self, + goal: Mapping[str, Any], + k: int, + strategy: dict | None = None, + ) -> RunResult: + """Run `k` independent branches in one round, promote the highest-fitness + survivor to `evolution/accepted`, demote the rest to `ledger/failed/`. + + `k <= 1` is delegated to `run_once` so single-branch callers retain the + exact byte-for-byte behavior covered by the v0.2 + phase-2 test suite. + Returned `RunResult.evaluation` carries the *summed* cost/tokens across + all k branches so hard-stop accounting in the CLI loop remains correct. + """ + if k <= 1: + return self.run_once(goal, strategy=strategy) + + self._ensure_git_repo() + self._ensure_accepted_branch() + baseline_commit = self._git("rev-parse", ACCEPTED_BRANCH) + + # Allocate k run_ids up front by materializing each run_dir before the + # next allocation, so _next_run_id observes the prior sibling. + run_ids: list[str] = [] + for _ in range(k): + rid = self._next_run_id() + (self.ledger_dir / "runs" / rid).mkdir(parents=True, exist_ok=False) + run_ids.append(rid) + + branches: list[dict] = [] + worktrees: list[Path] = [] + try: + for rid in run_ids: + br = self._run_single_branch(goal, rid, strategy, baseline_commit) + branches.append(br) + worktrees.append(br["worktree"]) + + winner_idx = self._select_winner(branches) + + total_cost = 0.0 + total_tokens = 0 + for br in branches: + ev = br["evaluation"] + try: + total_cost += float(ev.get("cost_usd") or 0.0) + except (TypeError, ValueError): + pass + try: + total_tokens += int(ev.get("tokens_used") or 0) + except (TypeError, ValueError): + pass + + winner_result: RunResult | None = None + for i, br in enumerate(branches): + is_winner = (winner_idx is not None and i == winner_idx) + if is_winner: + fitness = float(br["evaluation"].get("fitness", 0.0)) + decision = RunDecision( + accepted=True, + reason=f"k-branch winner: highest fitness {fitness:.4f}", + baseline_commit=baseline_commit, + candidate_commit=br["candidate_commit"], + rollback_target=baseline_commit, + ) + self._git("branch", "-f", ACCEPTED_BRANCH, br["candidate_commit"]) + else: + if br["scope_violation"]: + reason = f"scope_violation: {','.join(br['violations'])}" + elif not br["candidate_commit"]: + reason = "executor produced no repo changes" + elif winner_idx is None: + reason = "hard gates failed or evaluator rejected candidate" + else: + reason = f"k-branch: outranked by {branches[winner_idx]['run_id']}" + decision = RunDecision( + accepted=False, + reason=reason, + baseline_commit=baseline_commit, + candidate_commit=br["candidate_commit"], + rollback_target=baseline_commit, + ) + + self._write_json(br["run_dir"] / "decision.json", decision.__dict__) + plan_summary = "" + try: + plan_summary = str(self._read_json(br["run_dir"] / "plan.json").get("summary", "")) + except Exception: + pass + self._write_json( + br["run_dir"] / "reflection.json", + { + "run_id": br["run_id"], + "accepted": decision.accepted, + "reason": decision.reason, + "plan_summary": plan_summary, + "metrics": br["evaluation"].get("metrics", {}), + "fitness": float(br["evaluation"].get("fitness", 0.0)), + "created_at": self._now(), + }, + ) + if not decision.accepted: + failed_dir = self.ledger_dir / "failed" + failed_dir.mkdir(parents=True, exist_ok=True) + self._write_json(failed_dir / f"{br['run_id']}-summary.json", decision.__dict__) + if is_winner: + aggregated = dict(br["evaluation"]) + aggregated["cost_usd"] = total_cost + aggregated["tokens_used"] = total_tokens + winner_result = RunResult(br["run_id"], br["run_dir"], br["worktree"], decision, aggregated) + + self._record_accepted_commit() + + if winner_result is None: + # No branch passed hard gates: return the first branch as the + # round outcome (so CLI hard-stop bookkeeping still ticks), + # carrying the summed cost across all attempted branches. + br = branches[0] + decision_dict = self._read_json(br["run_dir"] / "decision.json") + aggregated = dict(br["evaluation"]) + aggregated["cost_usd"] = total_cost + aggregated["tokens_used"] = total_tokens + winner_result = RunResult( + br["run_id"], + br["run_dir"], + br["worktree"], + RunDecision(**decision_dict), + aggregated, + ) + return winner_result + finally: + for wt in worktrees: + if Path(wt).exists(): + try: + self._git("worktree", "remove", "--force", str(wt)) + except Exception: + pass + + def _run_single_branch( + self, + goal: Mapping[str, Any], + run_id: str, + strategy: dict | None, + baseline_commit: str, + ) -> dict: + """Run plan→execute→evaluate for one branch without promotion. + + The returned dict carries everything `run_once_parallel` needs to rank, + promote, and record a per-branch decision after all k branches finish. + """ + run_dir = self.ledger_dir / "runs" / run_id + worktree = self.ledger_dir / "worktrees" / run_id + worktree.parent.mkdir(parents=True, exist_ok=True) + branch = f"evolution/experiment/{run_id}" + self._git("worktree", "add", "-B", branch, str(worktree), baseline_commit) + + self._write_json(run_dir / "goal.json", dict(goal)) + if self.config_snapshot is not None: + self._write_json(run_dir / "config.json", self.config_snapshot) + + observation_path = run_dir / "observation.json" + observation = collect_observation(self.evidence_sources, self.target_repo) + write_observation(observation_path, observation) + + 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( + run_dir / "executor_input.json", + { + "run_id": run_id, + "goal": goal, + "baseline_commit": baseline_commit, + "plan_path": str(run_dir / "plan.json"), + "worktree": str(worktree), + }, + ) + self._run_role( + self.executor, + run_dir / "executor_input.json", + run_dir / "executor_output.json", + worktree, + ) + + candidate_commit = self._commit_candidate(worktree, run_id) + (run_dir / "candidate_commit.txt").write_text( + (candidate_commit or "") + "\n", encoding="utf-8" + ) + if candidate_commit: + patch = self._git_in(worktree, "diff", "--binary", baseline_commit, candidate_commit) + if patch and not patch.endswith("\n"): + patch += "\n" + else: + patch = "" + (run_dir / "patch.diff").write_text(patch, encoding="utf-8") + + scope_report = self._enforce_scope(worktree, baseline_commit, candidate_commit) + scope_violation = False + violations: list[str] = [] + if scope_report is not None and not scope_report.ok: + evaluation: dict = { + "hard_gates_passed": False, + "recommendation": "reject", + "reason": "scope_violation", + "violations": list(scope_report.violations), + "changed_files": list(scope_report.changed_files), + "allowed_paths": list(scope_report.allowed_paths), + "fitness": 0.0, + "metrics": {}, + } + self._write_json(run_dir / "evaluation.json", evaluation) + scope_violation = True + violations = list(scope_report.violations) + else: + self._write_json( + run_dir / "evaluator_input.json", + { + "run_id": run_id, + "goal": goal, + "baseline_commit": baseline_commit, + "candidate_commit": candidate_commit, + "patch_path": str(run_dir / "patch.diff"), + "worktree": str(worktree), + "observation_path": str(observation_path), + }, + ) + self._run_role( + self.evaluator, + run_dir / "evaluator_input.json", + run_dir / "evaluation.json", + worktree, + ) + evaluation = dict(self._read_json(run_dir / "evaluation.json")) + # Back-compat: synthesize a fitness when the evaluator omitted one. + if "fitness" not in evaluation: + evaluation["fitness"] = 1.0 if evaluation.get("hard_gates_passed") else 0.0 + + return { + "run_id": run_id, + "run_dir": run_dir, + "worktree": worktree, + "candidate_commit": candidate_commit, + "evaluation": evaluation, + "scope_violation": scope_violation, + "violations": violations, + } + + @staticmethod + def _select_winner(branches: Sequence[dict]) -> int | None: + """Pick the branch with the highest fitness among those that (1) produced + a commit, (2) survived scope check, (3) passed hard gates, and (4) the + evaluator recommended `accept` or `promote`. Ties broken by branch order + so behavior is deterministic for tests.""" + scored: list[tuple[float, int]] = [] + for i, br in enumerate(branches): + if not br["candidate_commit"]: + continue + if br["scope_violation"]: + continue + ev = br["evaluation"] + if not bool(ev.get("hard_gates_passed", False)): + continue + rec = str(ev.get("recommendation", "")).lower() + if rec not in {"accept", "promote"}: + continue + try: + fitness = float(ev.get("fitness", 0.0)) + except (TypeError, ValueError): + fitness = 0.0 + scored.append((fitness, i)) + if not scored: + return None + scored.sort(key=lambda pair: (-pair[0], pair[1])) + return scored[0][1] + def _build_history(self) -> list[dict]: """Scan ledger for past run reflections; return most recent N entries.""" runs_dir = self.ledger_dir / "runs" diff --git a/roles/evaluator.py b/roles/evaluator.py index 1219394..f575f7c 100755 --- a/roles/evaluator.py +++ b/roles/evaluator.py @@ -85,6 +85,8 @@ def main() -> None: Respond with ONLY a JSON object: - "hard_gates_passed": true if the change is safe and relevant, false otherwise - "recommendation": "accept" or "reject" +- "fitness": a float in [0.0, 1.0] — how strongly the change advances the goal (used by + k-branch parallel exploration to rank sibling branches) - "reason": one sentence explaining your decision - "metrics": {{}} (optional key/value metrics you can infer) """ @@ -114,6 +116,15 @@ def main() -> None: result.setdefault("recommendation", "reject") result.setdefault("reason", "") result.setdefault("metrics", {}) + # Back-compat: derive fitness from hard_gates_passed when the evaluator + # omitted it, so legacy evaluators keep working under k-branch ranking. + if "fitness" not in result: + result["fitness"] = 1.0 if result.get("hard_gates_passed") else 0.0 + else: + try: + result["fitness"] = float(result["fitness"]) + except (TypeError, ValueError): + result["fitness"] = 0.0 result["cost_usd"] = cost result["tokens_used"] = tokens diff --git a/tests/fixtures/evaluator_branch_fitness.py b/tests/fixtures/evaluator_branch_fitness.py new file mode 100644 index 0000000..3d1dd92 --- /dev/null +++ b/tests/fixtures/evaluator_branch_fitness.py @@ -0,0 +1,47 @@ +"""Evaluator fixture that reads the per-branch fitness hint set by the executor. + +The hint is the float written into EVOLUTION_MARKER.txt by executor_branch.py. +hard_gates_passed = (fitness > 0), so tests can mark a branch as "failing gates" +by setting EK_TEST_FITNESS_=0. +""" +from __future__ import annotations + +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() + +marker = Path(args.worktree) / "EVOLUTION_MARKER.txt" +fitness = 0.0 +if marker.exists(): + line = marker.read_text(encoding="utf-8").strip() + if line.startswith("fitness="): + try: + fitness = float(line.split("=", 1)[1]) + except ValueError: + fitness = 0.0 + +passes = fitness > 0 +Path(args.output).write_text( + json.dumps( + { + "hard_gates_passed": passes, + "recommendation": "promote" if passes else "reject", + "fitness": fitness, + "reason": f"fitness={fitness}", + "metrics": {"fitness": fitness}, + "cost_usd": 0.01, + "tokens_used": 100, + }, + indent=2, + sort_keys=True, + ) + + "\n", + encoding="utf-8", +) diff --git a/tests/fixtures/evaluator_src_fitness.py b/tests/fixtures/evaluator_src_fitness.py new file mode 100644 index 0000000..b54d57e --- /dev/null +++ b/tests/fixtures/evaluator_src_fitness.py @@ -0,0 +1,46 @@ +"""Evaluator fixture used with the partial-scope-violation parallel test. + +Looks for src/marker.txt (which executor_oob_for_run.py writes for in-scope +branches) and reads the fitness float from it. +""" +from __future__ import annotations + +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() + +marker = Path(args.worktree) / "src" / "marker.txt" +fitness = 0.0 +if marker.exists(): + line = marker.read_text(encoding="utf-8").strip() + if line.startswith("fitness="): + try: + fitness = float(line.split("=", 1)[1]) + except ValueError: + fitness = 0.0 + +passes = fitness > 0 +Path(args.output).write_text( + json.dumps( + { + "hard_gates_passed": passes, + "recommendation": "promote" if passes else "reject", + "fitness": fitness, + "reason": f"fitness={fitness}", + "metrics": {"fitness": fitness}, + "cost_usd": 0.02, + "tokens_used": 200, + }, + indent=2, + sort_keys=True, + ) + + "\n", + encoding="utf-8", +) diff --git a/tests/fixtures/executor_branch.py b/tests/fixtures/executor_branch.py new file mode 100644 index 0000000..6a64390 --- /dev/null +++ b/tests/fixtures/executor_branch.py @@ -0,0 +1,31 @@ +"""Executor fixture for k-branch tests. + +Writes EVOLUTION_MARKER.txt with a per-run "fitness hint" so a sibling evaluator +fixture can score branches differently within the same round. The hint comes +from an environment variable looked up by run_id, so each branch in a parallel +round can be assigned its own fitness from the test setup. +""" +from __future__ import annotations + +import argparse +import json +import os +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() + +payload = json.loads(Path(args.input).read_text(encoding="utf-8")) +run_id = payload["run_id"] +fitness = os.environ.get(f"EK_TEST_FITNESS_{run_id}", "0.5") + +worktree = Path(args.worktree) +(worktree / "EVOLUTION_MARKER.txt").write_text(f"fitness={fitness}\n", encoding="utf-8") +Path(args.output).write_text( + json.dumps({"changed": ["EVOLUTION_MARKER.txt"], "notes": f"branch {run_id}"}, indent=2) + "\n", + encoding="utf-8", +) diff --git a/tests/fixtures/executor_oob_for_run.py b/tests/fixtures/executor_oob_for_run.py new file mode 100644 index 0000000..5805905 --- /dev/null +++ b/tests/fixtures/executor_oob_for_run.py @@ -0,0 +1,37 @@ +"""Executor fixture that writes out-of-scope files for a specific run_id only. + +Used by the parallel scope-violation test: one branch violates scope while +others stay in-bounds. The "bad" run_id is selected via EK_TEST_OOB_RUN_ID. +""" +from __future__ import annotations + +import argparse +import json +import os +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() + +payload = json.loads(Path(args.input).read_text(encoding="utf-8")) +run_id = payload["run_id"] +worktree = Path(args.worktree) + +bad_run_id = os.environ.get("EK_TEST_OOB_RUN_ID", "") +if run_id == bad_run_id: + (worktree / "OUT_OF_SCOPE.txt").write_text("forbidden\n", encoding="utf-8") +else: + # Stay inside scope: write into allowed src/ subtree. + src = worktree / "src" + src.mkdir(exist_ok=True) + fitness = os.environ.get(f"EK_TEST_FITNESS_{run_id}", "0.5") + (src / "marker.txt").write_text(f"fitness={fitness}\n", encoding="utf-8") + +Path(args.output).write_text( + json.dumps({"changed": [], "notes": f"branch {run_id}"}, indent=2) + "\n", + encoding="utf-8", +) diff --git a/tests/fixtures/executor_unique_marker.py b/tests/fixtures/executor_unique_marker.py new file mode 100644 index 0000000..1959575 --- /dev/null +++ b/tests/fixtures/executor_unique_marker.py @@ -0,0 +1,24 @@ +"""Executor fixture that writes EVOLUTION_MARKER.txt with content unique per +run_id, so every branch produces a real diff against the baseline (otherwise +later rounds would re-emit the prior winner's content and commit nothing).""" +from __future__ import annotations + +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() + +payload = json.loads(Path(args.input).read_text(encoding="utf-8")) +run_id = payload["run_id"] +worktree = Path(args.worktree) +(worktree / "EVOLUTION_MARKER.txt").write_text(f"run={run_id}\n", encoding="utf-8") +Path(args.output).write_text( + json.dumps({"changed": ["EVOLUTION_MARKER.txt"], "notes": run_id}, indent=2) + "\n", + encoding="utf-8", +) diff --git a/tests/test_issue14.py b/tests/test_issue14.py new file mode 100644 index 0000000..22c0de8 --- /dev/null +++ b/tests/test_issue14.py @@ -0,0 +1,308 @@ +"""Tests for Issue #14: k-branch parallel exploration.""" +from __future__ import annotations + +import json +import os +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +from evolution_kernel.config import ConfigError, 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): + r = subprocess.run(["git", *args], cwd=cwd, text=True, capture_output=True, check=False) + if r.returncode != 0: + raise AssertionError(f"git {' '.join(args)} failed: {r.stderr}") + return r.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 +# --------------------------------------------------------------------------- + +class TestParallelConfig(unittest.TestCase): + + def test_default_k_is_one(self): + cfg = parse_config({"mission": "x"}) + self.assertEqual(cfg.parallel.k_branches, 1) + + def test_custom_k_parsed(self): + cfg = parse_config({"mission": "x", "parallel": {"k_branches": 4}}) + self.assertEqual(cfg.parallel.k_branches, 4) + + def test_k_must_be_positive(self): + with self.assertRaises(ConfigError): + parse_config({"mission": "x", "parallel": {"k_branches": 0}}) + + def test_k_must_be_int(self): + with self.assertRaises(ConfigError): + parse_config({"mission": "x", "parallel": {"k_branches": "two"}}) + + +# --------------------------------------------------------------------------- +# Governor — parallel core behavior +# --------------------------------------------------------------------------- + +class TestParallelGovernor(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) + self._saved_env: dict[str, str | None] = {} + + def tearDown(self): + for key, value in self._saved_env.items(): + if value is None: + os.environ.pop(key, None) + else: + os.environ[key] = value + self._tmp.cleanup() + + def _set_env(self, key: str, value: str) -> None: + self._saved_env.setdefault(key, os.environ.get(key)) + os.environ[key] = value + + def _gov( + self, + executor: str = "executor_branch.py", + evaluator: str = "evaluator_branch_fitness.py", + allowed_paths=(), + ) -> Governor: + return Governor( + target_repo=self.repo, + ledger_dir=self.ledger, + planner=_role("planner.py"), + executor=_role(executor), + evaluator=_role(evaluator), + allowed_paths=allowed_paths, + ) + + # ---- Done-when #3 coverage ---- + + def test_k_equal_one_matches_run_once(self): + """k=1 must delegate to run_once and return identical result shape.""" + gov = self._gov() + result = gov.run_once_parallel({"name": "t", "objective": "t"}, k=1) + # Single-branch run_id allocator returns "0001". + self.assertEqual(result.run_id, "0001") + self.assertTrue(result.decision.accepted) + # Only one run dir created. + runs = sorted((self.ledger / "runs").iterdir()) + self.assertEqual(len(runs), 1) + self.assertEqual(runs[0].name, "0001") + + def test_k_greater_than_one_creates_k_run_dirs(self): + """k=3 spawns three independent run dirs in one round, each with full + per-branch artifacts (planner_input, plan, evaluator_input, evaluation).""" + self._set_env("EK_TEST_FITNESS_0001", "0.2") + self._set_env("EK_TEST_FITNESS_0002", "0.9") + self._set_env("EK_TEST_FITNESS_0003", "0.5") + gov = self._gov() + gov.run_once_parallel({"name": "t", "objective": "t"}, k=3) + runs = sorted((self.ledger / "runs").iterdir()) + self.assertEqual([p.name for p in runs], ["0001", "0002", "0003"]) + for run_dir in runs: + for artifact in ( + "planner_input.json", "plan.json", + "executor_input.json", "executor_output.json", + "evaluation.json", "decision.json", "reflection.json", + ): + self.assertTrue((run_dir / artifact).exists(), + f"missing {artifact} in {run_dir.name}") + + def test_highest_fitness_branch_is_accepted(self): + """The branch with the highest fitness must be the one promoted.""" + self._set_env("EK_TEST_FITNESS_0001", "0.2") + self._set_env("EK_TEST_FITNESS_0002", "0.9") + self._set_env("EK_TEST_FITNESS_0003", "0.5") + gov = self._gov() + result = gov.run_once_parallel({"name": "t", "objective": "t"}, k=3) + self.assertEqual(result.run_id, "0002") + self.assertTrue(result.decision.accepted) + winner_decision = json.loads( + (self.ledger / "runs" / "0002" / "decision.json").read_text() + ) + self.assertTrue(winner_decision["accepted"]) + # evolution/accepted now points at the winner's candidate commit. + accepted_sha = _git(["rev-parse", ACCEPTED_BRANCH], self.repo) + self.assertEqual(accepted_sha, winner_decision["candidate_commit"]) + + def test_losing_branches_recorded_in_failed(self): + """Non-winning branches land in ledger/failed/ with a "outranked by" + reason that names the winner.""" + self._set_env("EK_TEST_FITNESS_0001", "0.2") + self._set_env("EK_TEST_FITNESS_0002", "0.9") + self._set_env("EK_TEST_FITNESS_0003", "0.5") + gov = self._gov() + gov.run_once_parallel({"name": "t", "objective": "t"}, k=3) + failed = self.ledger / "failed" + loser_files = sorted(p.name for p in failed.iterdir()) + self.assertEqual(loser_files, ["0001-summary.json", "0003-summary.json"]) + for fname in loser_files: + data = json.loads((failed / fname).read_text()) + self.assertFalse(data["accepted"]) + self.assertIn("outranked by 0002", data["reason"]) + + def test_all_worktrees_cleaned_up(self): + """Every per-branch worktree must be removed after the round, even + though their experiment branches stay around for audit.""" + self._set_env("EK_TEST_FITNESS_0001", "0.4") + self._set_env("EK_TEST_FITNESS_0002", "0.7") + self._set_env("EK_TEST_FITNESS_0003", "0.1") + gov = self._gov() + gov.run_once_parallel({"name": "t", "objective": "t"}, k=3) + worktree_root = self.ledger / "worktrees" + if worktree_root.exists(): + remaining = list(worktree_root.iterdir()) + self.assertEqual(remaining, [], + f"orphan worktrees: {[p.name for p in remaining]}") + # All three experiment branches should still exist (only the checkout + # is gone, not the audit trail). + branches_blob = _git(["branch", "--list", "evolution/experiment/*"], self.repo) + self.assertEqual( + sorted(line.strip().lstrip("* ") for line in branches_blob.splitlines() if line.strip()), + ["evolution/experiment/0001", "evolution/experiment/0002", "evolution/experiment/0003"], + ) + + def test_cost_and_tokens_summed_across_k_branches(self): + """The aggregated cost/tokens returned for hard-stop bookkeeping must be + the sum of all k per-branch evaluator reports.""" + self._set_env("EK_TEST_FITNESS_0001", "0.3") + self._set_env("EK_TEST_FITNESS_0002", "0.8") + self._set_env("EK_TEST_FITNESS_0003", "0.5") + gov = self._gov() + result = gov.run_once_parallel({"name": "t", "objective": "t"}, k=3) + # evaluator_branch_fitness.py reports cost=0.01 and tokens=100 per branch. + self.assertAlmostEqual(float(result.evaluation["cost_usd"]), 0.03, places=6) + self.assertEqual(int(result.evaluation["tokens_used"]), 300) + + def test_partial_scope_violation_does_not_block_other_branches(self): + """If one branch writes outside `allowed_paths` while siblings stay + in-scope, the violator is rejected and the highest-fitness in-scope + branch is still promoted normally.""" + self._set_env("EK_TEST_FITNESS_0001", "0.4") + self._set_env("EK_TEST_FITNESS_0002", "0.9") # violator + self._set_env("EK_TEST_FITNESS_0003", "0.6") + self._set_env("EK_TEST_OOB_RUN_ID", "0002") + gov = self._gov( + executor="executor_oob_for_run.py", + evaluator="evaluator_src_fitness.py", + allowed_paths=("src/",), + ) + result = gov.run_once_parallel({"name": "t", "objective": "t"}, k=3) + # 0003 has highest in-scope fitness (0.6); 0002's 0.9 was discarded. + self.assertEqual(result.run_id, "0003") + violator_eval = json.loads( + (self.ledger / "runs" / "0002" / "evaluation.json").read_text() + ) + self.assertEqual(violator_eval["reason"], "scope_violation") + violator_decision = json.loads( + (self.ledger / "failed" / "0002-summary.json").read_text() + ) + self.assertIn("scope_violation", violator_decision["reason"]) + + def test_no_branch_passes_means_accepted_unchanged(self): + """When every branch fails hard gates, the accepted branch must not + move and no winner is promoted.""" + # `evolution/accepted` is created lazily by the first run; track HEAD + # before instead, which is what the kernel seeds the accepted branch + # from on its very first invocation. + before = _git(["rev-parse", "HEAD"], self.repo) + self._set_env("EK_TEST_FITNESS_0001", "0") + self._set_env("EK_TEST_FITNESS_0002", "0") + self._set_env("EK_TEST_FITNESS_0003", "0") + gov = self._gov() + result = gov.run_once_parallel({"name": "t", "objective": "t"}, k=3) + self.assertFalse(result.decision.accepted) + after = _git(["rev-parse", ACCEPTED_BRANCH], self.repo) + self.assertEqual(before, after) + failed_files = sorted(p.name for p in (self.ledger / "failed").iterdir()) + self.assertEqual( + failed_files, + ["0001-summary.json", "0002-summary.json", "0003-summary.json"], + ) + + +# --------------------------------------------------------------------------- +# CLI integration — k=3 over 3 rounds (Done-when #4) +# --------------------------------------------------------------------------- + +class TestParallelCliLoop(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 test_k3_three_rounds_produces_9_runs_with_one_winner_each(self): + config_path = self.base / "evolution.yml" + config_path.write_text(f""" +mission: "parallel exploration" +parallel: + k_branches: 3 +hard_stops: + max_iterations: 3 + max_consecutive_failures: 5 +roles: + planner: ["python3", "{FIXTURES}/planner.py"] + executor: ["python3", "{FIXTURES}/executor_unique_marker.py"] + evaluator: ["python3", "{FIXTURES}/evaluator_accept.py"] +""") + from evolution_kernel.cli import main + rc = main([ + "--config", str(config_path), + "--repo", str(self.repo), + "--ledger", self.ledger, + "--loop", + ]) + # 3 rounds × k=3 = 9 run dirs, halted by max_iterations. + self.assertEqual(rc, 3) + runs = sorted((Path(self.ledger) / "runs").iterdir()) + self.assertEqual(len(runs), 9) + # Per round (3 sibling run_ids), exactly one accepted and two demoted. + accepted_count = 0 + for run_dir in runs: + decision = json.loads((run_dir / "decision.json").read_text()) + if decision["accepted"]: + accepted_count += 1 + self.assertEqual(accepted_count, 3) + # Per round, 2 demoted → 6 failed summaries total. + failed = list((Path(self.ledger) / "failed").iterdir()) + self.assertEqual(len(failed), 6) + + +if __name__ == "__main__": + unittest.main()