feat: goal evaluator + strategist (closes #10)

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())

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)

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(

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()