diff --git a/README.md b/README.md
index 4e8330d..d271571 100644
--- a/README.md
+++ b/README.md
@@ -1,242 +1,351 @@
# Evolution Kernel
- A general-purpose evolution engine for autonomously improving software projects.
+ Give an LLM a goal. Watch your repo improve itself. Stop when the budget runs out.
+
+
+
+ A ~1,200-line Python runtime for autonomous, multi-round code improvement — sandboxed, audited, and fully reversible.
中文
·
Protocol
- ·
- First Target
- 
- 
+
+ 
+
+ 
- 
+ 
-**Evolution Kernel** is a minimal protocol and runtime for autonomous, self-evolving software systems.
+---
-It is not a project-specific automation script. Its purpose is to make software evolution **controlled, reproducible, sandboxed, auditable, and reversible**. Any project can become an optimization target once it can expose a goal, a sandbox, and an evaluator.
+## What it does
-## Why It Exists
+Write a YAML file that says what "better" means. Evolution Kernel runs a tight loop:
-Modern coding agents can propose and modify code, but long-running software improvement needs more than code generation. It needs a kernel that can:
+1. **Observe** — collect the current metric (coverage %, benchmark score, lint count — whatever your shell command outputs)
+2. **Plan** — an LLM reads the metric and the history of prior attempts, then writes a concrete plan
+3. **Execute** — a coding agent (Aider or Claude Code) applies the plan inside a git worktree sandbox
+4. **Evaluate** — the evaluator re-runs your metric command and decides accept or reject
+5. **Commit or roll back** — accepted changes become a real git commit; rejected ones are discarded
+6. **Loop** — repeat until a budget limit fires (`max_iterations`, `max_total_usd`, `max_total_tokens`)
-- define what improvement means for a target project,
-- isolate each experiment before it touches the accepted branch,
-- evaluate candidate changes with repeatable criteria,
-- promote only accepted candidates,
-- keep a ledger of what happened and why.
+Every attempt — accepted or rejected — is written to a structured **ledger** so you can audit exactly what the LLM tried, what changed, and why each round was accepted or rejected.
-Evolution Kernel provides that loop as a small, inspectable runtime.
+---
-## Evolution Loop
+## Quick Start
-```mermaid
-flowchart LR
- Goal[Goal] --> Governor[Governor]
- Governor --> Planner[Planner]
- Planner --> Plan[plan.json]
- Plan --> Executor[Executor]
- Executor --> Candidate[Sandbox candidate]
- Candidate --> Evaluator[Evaluator]
- Evaluator --> Eval[evaluation.json]
- Eval --> Governor
- Governor --> Accepted[evolution/accepted]
- Governor --> Ledger[Ledger]
-```
+```bash
+# 1. Install (single runtime dependency: PyYAML)
+pip install evolution-kernel
-## First Optimization Target
+# 2. Write a goal config
+cat > evolution.yml << 'EOF'
+mission: "Increase src/ test coverage from 40% to 80%"
-Evolution Kernel is designed to optimize **any** software project. The first project being optimized is **Token-Ignition**, specifically its backend evaluator.
+evidence_sources:
+ - type: shell
+ command: >
+ python3 -m pytest --cov=src --cov-report=json -q &&
+ python3 -c "import json; d=json.load(open('coverage.json'));
+ print(f'coverage: {d[\"totals\"][\"percent_covered\"]:.1f}%')"
-Token-Ignition is therefore the first optimization target and reference adapter, not a hard dependency. It is used to prove that the kernel can safely and deterministically evolve a real codebase while keeping the runtime small.
+mutation_scope:
+ allowed_paths: ["tests/"]
-## Current Status
+hard_stops:
+ max_iterations: 20
+ max_consecutive_failures: 3
+ max_total_usd: 2.00
-The current v0 implementation provides the foundational runtime:
+llm:
+ provider: anthropic
+ model: claude-sonnet-4-6
+ api_key_env: ANTHROPIC_API_KEY
-| Area | What exists now |
-| --- | --- |
-| Governor | Deterministic orchestration for planning, execution, evaluation, promotion, rollback, and ledger updates. |
-| Sandbox | Git worktree-based experiment isolation. Candidate changes do not affect the accepted branch unless promoted. |
-| Role handoff | `planner`, `executor`, and `evaluator` run as isolated commands and communicate through JSON files. |
-| Promotion model | Accepted candidates advance the local `evolution/accepted` branch. Rejected experiments remain recorded but do not advance it. |
-| First adapter | A Token-Ignition adapter with a hand-written golden set for evaluator evolution. |
+coding_agent:
+ tool: aider
-## What It Does Not Do Yet
+roles:
+ planner: ["python3", "roles/planner.py"]
+ executor: ["bash", "roles/executor.sh"]
+ evaluator: ["python3", "roles/evaluator.py"]
+EOF
-| Not yet | Why it matters |
-| --- | --- |
-| LLM-native planner/executor | The current tests use fixture scripts; real agent integrations are the next step. |
-| Strong process/container sandboxing | Git worktrees isolate files, but executor and evaluator isolation should become stronger. |
-| Multi-target adapter framework | Token-Ignition is the first target; more adapters are needed to prove generality. |
-| Parallel evolution branches | v0 focuses on one accepted branch and a simple promotion path. |
+# 3. Run until the budget fires
+evolution-kernel --config evolution.yml --repo /path/to/your-project --ledger /tmp/ledger --loop
+```
-## Roadmap
+---
-- [ ] Add LLM-driven planner and executor implementations.
-- [ ] Add stronger sandbox isolation for executor and evaluator runs.
-- [ ] Generalize the adapter interface beyond Token-Ignition.
-- [ ] Add examples for multiple project types.
-- [ ] Support parallel evolution branches and richer merge strategies.
-- [ ] Improve reporting around ledger history, promotion decisions, and rejected candidates.
+## Example: raising test coverage from 40% to 80%
-## Documents
+The loop emits one JSON object per round. A realistic session looks like this:
-- [Protocol](docs/protocol.md)
-- [Token-Ignition First Task](docs/token-ignition-first-task.md)
+```
+Round 1 observe: coverage 40.2%
+ plan → "Add unit tests for src/parser.py — parse_tokens is completely uncovered"
+ execute → aider writes tests/test_parser.py (14 new assertions)
+ eval → coverage 51.7% — ACCEPT
+ commit → a3f1c9e "tests: cover parse_tokens (coverage 40→52%)"
+
+Round 2 observe: coverage 51.7%
+ plan → "Add edge-case tests for src/validator.py, missing branch coverage on error paths"
+ execute → aider extends tests/test_validator.py (+9 tests)
+ eval → coverage 63.4% — ACCEPT
+ commit → 8b2de01 "tests: validator edge cases (coverage 52→63%)"
+
+Round 3 observe: coverage 63.4%
+ plan → "Cover src/formatter.py — currently 0% covered"
+ execute → aider writes tests/test_formatter.py
+ eval → coverage 63.4% — new test file has wrong import path — REJECT
+ rollback → worktree discarded, main branch unchanged (consecutive_failures: 1)
+
+Round 4 observe: coverage 63.4%
+ plan → "tests/test_formatter.py failed due to import error; fix path and retry"
+ execute → aider fixes import in tests/test_formatter.py
+ eval → coverage 74.8% — ACCEPT
+ commit → 2c9af44 "tests: formatter coverage, fixed import (coverage 63→75%)"
+
+...
+
+Round 12 observe: coverage 80.1%
+ eval → coverage 80.1% — threshold reached — ACCEPT
+ commit → 9d7b321 "tests: final push past 80% target"
+
+{"halted": true, "reason": "max_iterations reached", "iterations": 20, "total_usd": 1.43, "total_tokens": 487201}
+```
-## Run Tests
+Each accepted change is a reversible git commit on the `evolution/accepted` branch. The LLM self-corrected on Round 4 using the rejection history from Round 3 — this is what history injection does.
-```bash
-python3 -m unittest discover -s tests -v
-python3 adapters/token_ignition/evaluate_golden_cases.py
-```
+---
-## CLI Shape
+## Ledger structure
-YAML-config mode (the primary MVP entry point — observer + scope + hard stops):
+Every round writes a full evidence trail. Nothing is stored in memory; an external auditor can reconstruct every decision from the ledger directory alone.
-```bash
-python3 -m evolution_kernel.cli \
- --config /path/to/evolution.yml \
- --repo /path/to/target-repo \
- --ledger /path/to/evolution-ledger
+```
+ledger/
+ .evolution_state.json # persisted counters (iterations, usd, tokens) — survives restarts
+ runs/
+ 0001/
+ config.json # full snapshot of your evolution.yml
+ observation.json # raw output of your evidence_sources commands
+ plan.json # LLM plan: summary, steps, expected_improvement
+ patch.diff # exact diff the executor applied
+ candidate_commit.txt # git SHA of the sandbox commit
+ evaluation.json # verdict + metrics + cost_usd + tokens_used
+ decision.json # accept / reject + reason
+ reflection.json # one-line summary injected into the next round's history
+ 0002/
+ ...
+ halted/
+ 20260501T120000Z.json # written when any hard stop fires
```
-Legacy direct-flags mode (still supported for the original golden-case tests):
+---
-```bash
-python3 -m evolution_kernel.cli \
- --repo /path/to/target-repo \
- --ledger /path/to/evolution-ledger \
- --goal /path/to/goal.json \
- --planner python3 /path/to/planner.py \
- --executor python3 /path/to/executor.py \
- --evaluator python3 /path/to/evaluator.py
+## Architecture
+
+```mermaid
+flowchart LR
+ Config[evolution.yml] --> Governor
+
+ subgraph loop ["Loop until hard stop"]
+ direction LR
+ Governor -->|"planner_input.json\n(goal + observation + history)"| Planner["Planner\nLLM"]
+ Planner -->|plan.json| Executor["Executor\nAider / Claude Code"]
+ Executor -->|patch in git worktree| Evaluator["Evaluator\nLLM + shell"]
+ Evaluator -->|evaluation.json| Governor
+ end
+
+ Governor -->|"accept → git commit"| AcceptedBranch[evolution/accepted]
+ Governor -->|"reject → discard worktree"| Ledger[Ledger]
+ Governor --> Ledger
```
-Reset the persistent hard-stop state (after a halt) without running a loop:
+**The Governor is intentionally dumb.** It is pure orchestration — no LLM calls of its own. All intelligence lives in the three role scripts. You can swap any role for your own implementation; the Governor only cares about the JSON files roles read and write.
-```bash
-python3 -m evolution_kernel.cli --reset --ledger /path/to/evolution-ledger
-```
+**Roles communicate through files, not shared memory.** The planner never talks directly to the executor. The evaluator never sees the executor's self-assessment. The only shared state is the ledger.
-Each role command receives:
+---
-```text
---input
---output
---worktree
-```
+## Capabilities
-## MVP Usage (closed loop with observer, scope, hard stops)
+| Feature | Status |
+|---|---|
+| Multi-round LLM loop with memory (history injection) | ✅ Working |
+| Budget guards: `max_total_usd`, `max_total_tokens` | ✅ Working |
+| Iteration / consecutive-failure hard stops | ✅ Working |
+| Full ledger audit trail (survives process restarts) | ✅ Working |
+| Git worktree sandbox — every attempt isolated | ✅ Working |
+| Scope enforcement — rejects changes outside `allowed_paths` | ✅ Working |
+| Config-driven: swap LLM provider, model, coding agent | ✅ Working |
+| Aider and Claude Code executor support | ✅ Working |
+| Anthropic and OpenAI planner/evaluator support | ✅ Working |
+| Goal evaluator — stops when mission is "won" | 🔧 PR #5 |
+| k-branch parallel exploration (FunSearch style) | 🔧 PR #6 |
+| Process sandbox (firejail / bwrap) for production safety | 🔧 PR #7 |
-This MVP wires the full closed loop described in the protocol:
-`config -> observe -> plan/execute -> evaluate -> accept/reject -> ledger`.
+---
-### 1. Author an `evolution.yml`
+## Configuration reference
```yaml
-mission: "Add a minimal in-scope mutation so the evaluator accepts."
+# Required — free-text statement of what "better" means
+mission: "Increase src/ test coverage from 40% to 80%"
+# How to measure the current state of the target repo
evidence_sources:
- - type: file
- path: metrics.json
- - type: shell
- command: "bash scripts/status.sh"
+ - type: shell # runs a command; stdout goes into observation.json
+ command: "python3 -m pytest --cov=src -q && ..."
+ - type: file # reads a file; content goes into observation.json
+ path: "metrics.json"
+# Only files under these paths may be modified by the executor
mutation_scope:
allowed_paths:
- - "src/"
+ - "tests/" # changes outside this list are auto-rejected
+# When to stop
hard_stops:
- max_iterations: 3
- max_consecutive_failures: 2
-
+ max_iterations: 10 # total rounds (required, must be ≥ 1)
+ max_consecutive_failures: 3 # consecutive rejections before halt (required)
+ max_total_usd: 0.0 # 0 = unlimited
+ max_total_tokens: 0 # 0 = unlimited
+
+# LLM used by the planner and evaluator role scripts
+llm:
+ provider: anthropic # anthropic | openai
+ model: claude-sonnet-4-6
+ api_key_env: ANTHROPIC_API_KEY
+
+# Coding agent used by the executor role script
+coding_agent:
+ tool: aider # aider | claude-code
+
+# How many past rounds the planner sees as context
+history:
+ max_entries: 10
+
+# The three role commands (each receives --input, --output, --worktree)
roles:
- planner: ["python3", "bots/planner.py"]
- executor: ["python3", "bots/executor.py"]
- evaluator: ["python3", "bots/evaluator.py"]
+ planner: ["python3", "roles/planner.py"]
+ executor: ["bash", "roles/executor.sh"]
+ evaluator: ["python3", "roles/evaluator.py"]
+```
+
+**Switch to OpenAI:**
+
+```yaml
+llm:
+ provider: openai
+ model: gpt-4o
+ api_key_env: OPENAI_API_KEY
```
-`evidence_sources` are read into `observation.json` before the planner runs.
-`mutation_scope.allowed_paths` are enforced after the executor commits — anything
-outside the scope is auto-rejected with `decision.reason = "scope_violation: ..."`.
-`hard_stops` persist across runs in `/.evolution_state.json` so a stuck
-loop halts even across CLI invocations.
+**Switch to Claude Code as coding agent:**
-### 2. Run a single iteration
+```yaml
+coding_agent:
+ tool: claude-code
+```
+
+---
+
+## CLI reference
```bash
-# one-time: install the package (pulls PyYAML, the only runtime dep)
-python3 -m pip install -e .
+# Run the multi-round loop (recommended — stops when a hard stop fires)
+evolution-kernel --config evolution.yml --repo /path/to/repo --ledger /tmp/ledger --loop
-# one-time: prepare a target repo
-bash examples/demo_target/setup.sh
+# Run exactly one round
+evolution-kernel --config evolution.yml --repo /path/to/repo --ledger /tmp/ledger
-python3 -m evolution_kernel.cli \
- --config examples/evolution.yml \
- --repo examples/demo_target \
- --ledger /tmp/ek-ledger
+# Reset hard-stop counters to start a fresh session
+evolution-kernel --ledger /tmp/ledger --reset
```
-> The `pip install -e .` step is only needed once per environment — it pulls
-> `PyYAML>=6.0` (declared in `pyproject.toml`). After that the three-line
-> command above is reproducible from a clean checkout.
+Exit codes: `0` = clean finish, `3` = halted by a hard stop.
+
+---
+
+## Install
+
+```bash
+pip install evolution-kernel
+```
-Reset the persistent hard-stop counters when you want to start fresh:
+From source (the only runtime dependency is PyYAML):
```bash
-python3 -m evolution_kernel.cli --reset --ledger /tmp/ek-ledger
+git clone https://github.com/Protocol-zero-0/evolution-kernel.git
+cd evolution-kernel
+pip install -e .
```
-### 3. Inspect the ledger
+Python 3.10 or later required.
+
+---
+
+## Running the tests
+
+```bash
+python3 -m pytest tests/ -v
+```
+
+All tests run locally with no network calls — roles are replaced by lightweight fixture scripts.
+
+---
+
+## Writing your own role scripts
-Every run produces a directory under `/runs//` containing the
-full evidence trail:
+Each role is an executable that receives three arguments:
```text
-goal.json # legacy mode only
-config.json # full snapshot of the YAML config (full mode)
-observation.json # what the observer collected before planning
-plan.json # planner output
-patch.diff # diff between baseline and candidate commit
-candidate_commit.txt # the candidate commit hash inside the sandbox
-evaluation.json # evaluator output (synthesised on scope_violation)
-decision.json # accept / reject + reason
-reflection.json # post-decision summary
-```
-
-### 4. Acceptance criteria -> tests
-
-The six acceptance bullets from issue #1 each map to a test in
-`tests/test_acceptance.py`:
-
-| # | Acceptance bullet | Test |
-| - | --- | --- |
-| 1 | Accept advances `evolution/accepted` | `test_accept_advances_accepted_branch` |
-| 2 | Reject does not advance it | `test_reject_does_not_advance_accepted_branch` |
-| 3 | Mutation scope enforced + violation logged | `test_scope_violation_is_rejected_and_logged` |
-| 4 | Observer writes `observation.json` (file + shell) | `test_observer_writes_observation_with_file_and_shell` |
-| 5 | Hard stops halt then `reset` re-enables | `test_hard_stop_blocks_then_reset_allows_via_cli` |
-| 6 | Ledger contains all required artifacts | `test_ledger_contains_all_required_artifacts` |
-
-### What this MVP intentionally does **not** do
-
-In line with the issue's "out of scope" list:
-
-- No LLM / agent-swarm / dashboard.
-- No PR router and no auto-merge to upstream `main`.
-- No multi-target adapter framework — the only example target is
- `examples/demo_target/`.
-- No container/process sandbox beyond git worktrees.
-
-These are the natural next steps once the kernel itself is trusted.
+--input JSON the governor prepared for this role
+--output JSON the role must write before exiting
+--worktree path to the isolated git sandbox checkout
+```
+
+The built-in `roles/planner.py`, `roles/executor.sh`, and `roles/evaluator.py` are the reference implementation. Copy and modify them, or replace them entirely with a shell script, a Python program, or a Docker call. The Governor has no opinion on what runs inside a role.
+
+---
+
+## Rollback
+
+Every accepted change is a commit on the `evolution/accepted` branch. To undo everything from a session:
+
+```bash
+git checkout evolution/accepted
+git log --oneline # find the baseline commit before the session
+git reset --hard # roll back all accepted changes
+```
+
+Rejected experiments are never promoted, so only the changes your evaluator explicitly accepted survive.
+
+---
+
+## Project layout
+
+```
+evolution_kernel/ # ~1,200-line runtime (Governor, Observer, HardStops, Config, CLI)
+roles/ # reference planner, executor, and evaluator implementations
+examples/ # demo target + evolution.yml to run out of the box
+docs/ # protocol spec
+tests/ # unit + acceptance tests (39 tests, no network required)
+```
+
+---
+
+## License
+
+MIT. See [LICENSE](LICENSE).
diff --git a/evolution_kernel/cli.py b/evolution_kernel/cli.py
index 72f6895..51b01b2 100644
--- a/evolution_kernel/cli.py
+++ b/evolution_kernel/cli.py
@@ -1,18 +1,15 @@
"""Evolution Kernel command-line entry point.
-The CLI shape mirrors the suggested form in the project's MVP brief:
+Usage:
- python -m evolution_kernel.cli \
- --config examples/evolution.yml \
- --repo /path/to/target-repo \
- --ledger /tmp/evolution-ledger
+ # Run once:
+ evolution-kernel --config examples/evolution.yml --repo /path/to/repo --ledger /tmp/ledger
-Two extra modes are supported alongside this primary form:
+ # Run until hard stops trigger (multi-round loop):
+ evolution-kernel --config examples/evolution.yml --repo /path/to/repo --ledger /tmp/ledger --loop
-* ``--goal goal.json`` runs the legacy direct-flags loop (no observer / scope /
- hard-stops) so the original golden-case tests keep working unchanged.
-* ``--reset`` clears the persisted hard-stop state for the given ledger and
- exits — used to re-enable a halted loop after a human review.
+ # Reset hard-stop state:
+ evolution-kernel --ledger /tmp/ledger --reset
"""
from __future__ import annotations
@@ -32,7 +29,7 @@
def main(argv: Sequence[str] | None = None) -> int:
parser = argparse.ArgumentParser(
prog="evolution-kernel",
- description="Run one Evolution Kernel experiment under MVP constraints.",
+ description="Run Evolution Kernel experiments.",
)
parser.add_argument("--repo", help="Target git repository (required unless --reset)")
parser.add_argument("--ledger", required=True, help="Ledger directory")
@@ -43,6 +40,7 @@ def main(argv: Sequence[str] | None = None) -> int:
parser.add_argument("--executor", nargs="+", help="Executor argv (overrides config.roles.executor)")
parser.add_argument("--evaluator", nargs="+", help="Evaluator argv (overrides config.roles.evaluator)")
parser.add_argument("--run-id", default=None)
+ parser.add_argument("--loop", action="store_true", help="Run until hard stops trigger (multi-round).")
parser.add_argument(
"--reset",
action="store_true",
@@ -75,7 +73,7 @@ def _cmd_reset(args: argparse.Namespace) -> int:
return 0
-def _run_with_config(args: argparse.Namespace, cfg: EvolutionConfig) -> int:
+def _make_governor(args: argparse.Namespace, cfg: EvolutionConfig) -> Governor:
planner = tuple(args.planner) if args.planner else cfg.roles.planner
executor = tuple(args.executor) if args.executor else cfg.roles.executor
evaluator = tuple(args.evaluator) if args.evaluator else cfg.roles.evaluator
@@ -84,44 +82,102 @@ def _run_with_config(args: argparse.Namespace, cfg: EvolutionConfig) -> int:
"error: planner/executor/evaluator must be defined in config.roles or via flags",
file=sys.stderr,
)
- return 2
+ raise SystemExit(2)
+ return Governor(
+ target_repo=args.repo,
+ ledger_dir=args.ledger,
+ planner=RoleCommand(list(planner)),
+ executor=RoleCommand(list(executor)),
+ evaluator=RoleCommand(list(evaluator)),
+ evidence_sources=cfg.evidence_sources,
+ allowed_paths=cfg.mutation_scope.allowed_paths,
+ config_snapshot=cfg.raw,
+ history_max_entries=cfg.history.max_entries,
+ )
+
+
+def _run_with_config(args: argparse.Namespace, cfg: EvolutionConfig) -> int:
+ try:
+ governor = _make_governor(args, cfg)
+ except SystemExit as e:
+ return int(e.code)
+
+ goal = {"name": cfg.mission, "objective": cfg.mission}
+
+ if args.loop:
+ return _run_loop(args, cfg, governor, goal)
+ # Single run
state = hard_stops.load_state(args.ledger)
allowed, why = hard_stops.precheck(
state,
cfg.hard_stops.max_iterations,
cfg.hard_stops.max_consecutive_failures,
+ max_total_usd=cfg.hard_stops.max_total_usd,
+ max_total_tokens=cfg.hard_stops.max_total_tokens,
)
if not allowed:
- # Even when blocked, leave an audit record so the ledger covers every
- # invocation, not just the ones that actually ran the loop.
_record_halted(args.ledger, state, why)
print(json.dumps({"halted": True, "reason": why}, indent=2, sort_keys=True))
return 3
- goal = {"name": cfg.mission, "objective": cfg.mission}
- governor = Governor(
- target_repo=args.repo,
- ledger_dir=args.ledger,
- planner=RoleCommand(list(planner)),
- executor=RoleCommand(list(executor)),
- evaluator=RoleCommand(list(evaluator)),
- evidence_sources=cfg.evidence_sources,
- allowed_paths=cfg.mutation_scope.allowed_paths,
- config_snapshot=cfg.raw,
- )
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,
accepted=result.decision.accepted,
max_iterations=cfg.hard_stops.max_iterations,
max_consecutive_failures=cfg.hard_stops.max_consecutive_failures,
+ cost_usd=cost_usd,
+ tokens_used=tokens_used,
+ max_total_usd=cfg.hard_stops.max_total_usd,
+ max_total_tokens=cfg.hard_stops.max_total_tokens,
)
hard_stops.save_state(args.ledger, new_state)
_print_result(result, halted=new_state.halted, halt_reason=new_state.halt_reason)
return 0
+def _run_loop(
+ args: argparse.Namespace,
+ cfg: EvolutionConfig,
+ governor: Governor,
+ goal: dict,
+) -> int:
+ """Run until hard stops trigger. Each iteration saves state immediately."""
+ while True:
+ state = hard_stops.load_state(args.ledger)
+ allowed, why = hard_stops.precheck(
+ state,
+ cfg.hard_stops.max_iterations,
+ cfg.hard_stops.max_consecutive_failures,
+ max_total_usd=cfg.hard_stops.max_total_usd,
+ max_total_tokens=cfg.hard_stops.max_total_tokens,
+ )
+ if not allowed:
+ _record_halted(args.ledger, state, why)
+ print(json.dumps({"halted": True, "reason": why}, indent=2, sort_keys=True))
+ return 3
+
+ result = governor.run_once(goal)
+ cost_usd, tokens_used = _safe_cost(result.evaluation)
+ new_state = hard_stops.record_outcome(
+ state,
+ accepted=result.decision.accepted,
+ max_iterations=cfg.hard_stops.max_iterations,
+ max_consecutive_failures=cfg.hard_stops.max_consecutive_failures,
+ cost_usd=cost_usd,
+ tokens_used=tokens_used,
+ max_total_usd=cfg.hard_stops.max_total_usd,
+ max_total_tokens=cfg.hard_stops.max_total_tokens,
+ )
+ hard_stops.save_state(args.ledger, new_state)
+ _print_result(result, halted=new_state.halted, halt_reason=new_state.halt_reason)
+ if new_state.halted:
+ _record_halted(args.ledger, new_state, new_state.halt_reason)
+ return 3
+
+
def _run_legacy(args: argparse.Namespace) -> int:
if not (args.planner and args.executor and args.evaluator):
print(
@@ -142,6 +198,19 @@ def _run_legacy(args: argparse.Namespace) -> int:
return 0
+def _safe_cost(evaluation: dict) -> tuple[float, int]:
+ """Extract cost fields defensively; return (0.0, 0) on any parse error."""
+ try:
+ cost_usd = float(evaluation.get("cost_usd") or 0.0)
+ except (TypeError, ValueError):
+ cost_usd = 0.0
+ try:
+ tokens_used = int(evaluation.get("tokens_used") or 0)
+ except (TypeError, ValueError):
+ tokens_used = 0
+ return cost_usd, tokens_used
+
+
def _record_halted(
ledger_dir: str,
state: hard_stops.HardStopState,
@@ -155,8 +224,9 @@ def _record_halted(
"reason": reason,
"iterations": state.iterations,
"consecutive_failures": state.consecutive_failures,
+ "total_usd": state.total_usd,
+ "total_tokens": state.total_tokens,
}
- # Suffix with sequence number to avoid collisions within the same second.
base = halted_dir / f"{ts}.json"
target = base
n = 1
diff --git a/evolution_kernel/config.py b/evolution_kernel/config.py
index 97ab5a3..03651df 100644
--- a/evolution_kernel/config.py
+++ b/evolution_kernel/config.py
@@ -4,6 +4,17 @@
mission: "free-text statement of intent"
+ llm:
+ provider: anthropic # anthropic | openai
+ model: claude-sonnet-4-6
+ api_key_env: ANTHROPIC_API_KEY
+
+ coding_agent:
+ tool: aider # aider | claude-code
+
+ history:
+ max_entries: 10
+
evidence_sources:
- type: file
path: "./metrics.json"
@@ -16,8 +27,10 @@
- "tests/"
hard_stops:
- max_iterations: 3
- max_consecutive_failures: 2
+ max_iterations: 10
+ max_consecutive_failures: 3
+ max_total_usd: 1.00 # 0.0 = unlimited
+ max_total_tokens: 500000 # 0 = unlimited
Validation prefers human-readable errors over raw tracebacks so that bad configs
can be fixed without reading source.
@@ -52,6 +65,8 @@ class MutationScope:
class HardStops:
max_iterations: int = 1
max_consecutive_failures: int = 1
+ max_total_usd: float = 0.0 # 0.0 = unlimited
+ max_total_tokens: int = 0 # 0 = unlimited
@dataclass(frozen=True)
@@ -61,6 +76,23 @@ class Roles:
evaluator: tuple[str, ...] = ()
+@dataclass(frozen=True)
+class LLMConfig:
+ provider: str = "anthropic" # anthropic | openai
+ model: str = "claude-sonnet-4-6"
+ api_key_env: str = "ANTHROPIC_API_KEY"
+
+
+@dataclass(frozen=True)
+class CodingAgentConfig:
+ tool: str = "aider" # aider | claude-code
+
+
+@dataclass(frozen=True)
+class HistoryConfig:
+ max_entries: int = 10
+
+
@dataclass(frozen=True)
class EvolutionConfig:
mission: str
@@ -68,6 +100,9 @@ class EvolutionConfig:
mutation_scope: MutationScope = field(default_factory=MutationScope)
hard_stops: HardStops = field(default_factory=HardStops)
roles: Roles = field(default_factory=Roles)
+ llm: LLMConfig = field(default_factory=LLMConfig)
+ coding_agent: CodingAgentConfig = field(default_factory=CodingAgentConfig)
+ history: HistoryConfig = field(default_factory=HistoryConfig)
raw: Mapping[str, Any] = field(default_factory=dict)
@@ -97,6 +132,9 @@ def parse_config(raw: Mapping[str, Any]) -> EvolutionConfig:
mutation_scope = _parse_mutation_scope(raw.get("mutation_scope", {}))
hard_stops = _parse_hard_stops(raw.get("hard_stops", {}))
roles = _parse_roles(raw.get("roles", {}))
+ llm = _parse_llm(raw.get("llm", {}))
+ coding_agent = _parse_coding_agent(raw.get("coding_agent", {}))
+ history = _parse_history(raw.get("history", {}))
return EvolutionConfig(
mission=mission.strip(),
@@ -104,6 +142,9 @@ def parse_config(raw: Mapping[str, Any]) -> EvolutionConfig:
mutation_scope=mutation_scope,
hard_stops=hard_stops,
roles=roles,
+ llm=llm,
+ coding_agent=coding_agent,
+ history=history,
raw=dict(raw),
)
@@ -182,4 +223,53 @@ def _parse_hard_stops(value: Any) -> HardStops:
for label, n in (("max_iterations", max_iterations), ("max_consecutive_failures", max_failures)):
if not isinstance(n, int) or isinstance(n, bool) or n < 1:
raise ConfigError(f"`hard_stops.{label}` must be a positive integer, got {n!r}")
- return HardStops(max_iterations=max_iterations, max_consecutive_failures=max_failures)
+ usd_raw = value.get("max_total_usd", 0.0)
+ tok_raw = value.get("max_total_tokens", 0)
+ try:
+ max_total_usd = float(usd_raw)
+ except (TypeError, ValueError):
+ raise ConfigError(f"`hard_stops.max_total_usd` must be a number, got {usd_raw!r}")
+ try:
+ max_total_tokens = int(tok_raw)
+ except (TypeError, ValueError):
+ raise ConfigError(f"`hard_stops.max_total_tokens` must be an integer, got {tok_raw!r}")
+ if max_total_usd < 0:
+ raise ConfigError("`hard_stops.max_total_usd` must be >= 0")
+ if max_total_tokens < 0:
+ raise ConfigError("`hard_stops.max_total_tokens` must be >= 0")
+ return HardStops(
+ max_iterations=max_iterations,
+ max_consecutive_failures=max_failures,
+ max_total_usd=max_total_usd,
+ max_total_tokens=max_total_tokens,
+ )
+
+
+def _parse_llm(value: Any) -> LLMConfig:
+ if not isinstance(value, Mapping):
+ raise ConfigError("`llm` must be a mapping")
+ provider = value.get("provider", "anthropic")
+ model = value.get("model", "claude-sonnet-4-6")
+ api_key_env = value.get("api_key_env", "ANTHROPIC_API_KEY")
+ for label, v in (("provider", provider), ("model", model), ("api_key_env", api_key_env)):
+ if not isinstance(v, str) or not v.strip():
+ raise ConfigError(f"`llm.{label}` must be a non-empty string")
+ return LLMConfig(provider=provider.strip(), model=model.strip(), api_key_env=api_key_env.strip())
+
+
+def _parse_coding_agent(value: Any) -> CodingAgentConfig:
+ if not isinstance(value, Mapping):
+ raise ConfigError("`coding_agent` must be a mapping")
+ tool = value.get("tool", "aider")
+ if not isinstance(tool, str) or not tool.strip():
+ raise ConfigError("`coding_agent.tool` must be a non-empty string")
+ return CodingAgentConfig(tool=tool.strip())
+
+
+def _parse_history(value: Any) -> HistoryConfig:
+ if not isinstance(value, Mapping):
+ raise ConfigError("`history` must be a mapping")
+ max_entries = value.get("max_entries", 10)
+ 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)
diff --git a/evolution_kernel/governor.py b/evolution_kernel/governor.py
index b0ce49d..1583261 100644
--- a/evolution_kernel/governor.py
+++ b/evolution_kernel/governor.py
@@ -52,6 +52,7 @@ def __init__(
evidence_sources: Sequence[EvidenceSource] = (),
allowed_paths: Sequence[str] = (),
config_snapshot: Mapping[str, Any] | None = None,
+ history_max_entries: int = 10,
) -> None:
self.target_repo = Path(target_repo).resolve()
self.ledger_dir = Path(ledger_dir).resolve()
@@ -61,6 +62,7 @@ def __init__(
self.evidence_sources = tuple(evidence_sources)
self.allowed_paths = tuple(allowed_paths)
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:
self._ensure_git_repo()
@@ -96,6 +98,7 @@ def run_once(self, goal: Mapping[str, Any], run_id: str | None = None) -> RunRes
"ledger_dir": str(self.ledger_dir),
"observation_path": str(observation_path),
"allowed_paths": list(self.allowed_paths),
+ "history": self._build_history(),
},
)
self._run_role(self.planner, run_dir / "planner_input.json", run_dir / "plan.json", worktree)
@@ -182,12 +185,20 @@ def run_once(self, goal: Mapping[str, Any], run_id: str | None = None) -> RunRes
self._git("branch", "-f", ACCEPTED_BRANCH, candidate_commit)
self._record_accepted_commit()
+ # Pull plan summary for history — more informative than decision.reason.
+ plan_summary = ""
+ try:
+ plan_data = self._read_json(run_dir / "plan.json")
+ plan_summary = str(plan_data.get("summary", ""))
+ except Exception:
+ pass
self._write_json(
run_dir / "reflection.json",
{
"run_id": run_id,
"accepted": decision.accepted,
"reason": decision.reason,
+ "plan_summary": plan_summary,
"metrics": evaluation.get("metrics", {}),
"created_at": self._now(),
},
@@ -203,6 +214,28 @@ def run_once(self, goal: Mapping[str, Any], run_id: str | None = None) -> RunRes
if worktree.exists():
self._git("worktree", "remove", "--force", str(worktree))
+ def _build_history(self) -> list[dict]:
+ """Scan ledger for past run reflections; return most recent N entries."""
+ runs_dir = self.ledger_dir / "runs"
+ if not runs_dir.exists():
+ return []
+ entries = []
+ for run_dir in sorted(runs_dir.iterdir()):
+ reflection = run_dir / "reflection.json"
+ if not reflection.exists():
+ continue
+ try:
+ data = self._read_json(reflection)
+ entries.append({
+ "run_id": data.get("run_id", run_dir.name),
+ "accepted": data.get("accepted", False),
+ "summary": data.get("plan_summary") or data.get("reason", ""),
+ "metrics": data.get("metrics", {}),
+ })
+ except Exception:
+ pass
+ return entries[-self.history_max_entries:]
+
def _decide(
self,
evaluation: Mapping[str, Any],
diff --git a/evolution_kernel/hard_stops.py b/evolution_kernel/hard_stops.py
index 493df79..e26e57c 100644
--- a/evolution_kernel/hard_stops.py
+++ b/evolution_kernel/hard_stops.py
@@ -20,6 +20,8 @@
class HardStopState:
iterations: int = 0
consecutive_failures: int = 0
+ total_usd: float = 0.0
+ total_tokens: int = 0
halted: bool = False
halt_reason: str | None = None
@@ -31,6 +33,8 @@ def from_json(cls, data: Mapping[str, Any]) -> "HardStopState":
return cls(
iterations=int(data.get("iterations", 0)),
consecutive_failures=int(data.get("consecutive_failures", 0)),
+ total_usd=float(data.get("total_usd", 0.0)),
+ total_tokens=int(data.get("total_tokens", 0)),
halted=bool(data.get("halted", False)),
halt_reason=data.get("halt_reason"),
)
@@ -63,7 +67,14 @@ def save_state(ledger_dir: Path | str, state: HardStopState) -> None:
os.replace(tmp, p)
-def precheck(state: HardStopState, max_iterations: int, max_consecutive_failures: int) -> tuple[bool, str | None]:
+def precheck(
+ state: HardStopState,
+ max_iterations: int,
+ max_consecutive_failures: int,
+ *,
+ max_total_usd: float = 0.0,
+ max_total_tokens: int = 0,
+) -> tuple[bool, str | None]:
"""Return (allowed, reason). reason is None when allowed."""
if state.halted:
return False, state.halt_reason or "halted"
@@ -71,6 +82,10 @@ def precheck(state: HardStopState, max_iterations: int, max_consecutive_failures
return False, f"max_iterations reached ({max_iterations})"
if state.consecutive_failures >= max_consecutive_failures:
return False, f"max_consecutive_failures reached ({max_consecutive_failures})"
+ if max_total_usd > 0 and state.total_usd >= max_total_usd:
+ return False, f"max_total_usd reached ({max_total_usd})"
+ if max_total_tokens > 0 and state.total_tokens >= max_total_tokens:
+ return False, f"max_total_tokens reached ({max_total_tokens})"
return True, None
@@ -80,9 +95,15 @@ def record_outcome(
accepted: bool,
max_iterations: int,
max_consecutive_failures: int,
+ cost_usd: float = 0.0,
+ tokens_used: int = 0,
+ max_total_usd: float = 0.0,
+ max_total_tokens: int = 0,
) -> HardStopState:
"""Update counters after a run; mark halted if any limit just tripped."""
state.iterations += 1
+ state.total_usd += cost_usd
+ state.total_tokens += tokens_used
if accepted:
state.consecutive_failures = 0
else:
@@ -93,6 +114,12 @@ def record_outcome(
elif state.consecutive_failures >= max_consecutive_failures:
state.halted = True
state.halt_reason = f"max_consecutive_failures reached ({max_consecutive_failures})"
+ elif max_total_usd > 0 and state.total_usd >= max_total_usd:
+ state.halted = True
+ state.halt_reason = f"max_total_usd reached ({max_total_usd:.4f})"
+ elif max_total_tokens > 0 and state.total_tokens >= max_total_tokens:
+ state.halted = True
+ state.halt_reason = f"max_total_tokens reached ({max_total_tokens})"
return state
diff --git a/examples/evolution.yml b/examples/evolution.yml
index eceedbc..8abf941 100644
--- a/examples/evolution.yml
+++ b/examples/evolution.yml
@@ -1,5 +1,19 @@
mission: "Improve the demo target so its evaluator passes a simple metric check, under strict reproducibility constraints."
+# LLM configuration — all role scripts read this via config.json in the run dir.
+llm:
+ provider: anthropic # anthropic | openai
+ model: claude-sonnet-4-6
+ api_key_env: ANTHROPIC_API_KEY # name of the env var holding the key
+
+# Coding agent used by roles/executor.sh
+coding_agent:
+ tool: aider # aider | claude-code
+
+# How many past run reflections to inject into each planner call
+history:
+ max_entries: 10
+
evidence_sources:
- type: file
path: "metrics.json"
@@ -11,10 +25,12 @@ mutation_scope:
- "src/"
hard_stops:
- max_iterations: 3
- max_consecutive_failures: 2
+ max_iterations: 10
+ max_consecutive_failures: 3
+ max_total_usd: 1.00 # stop if total LLM spend reaches $1
+ max_total_tokens: 500000 # stop if total tokens reaches 500k
roles:
- planner: ["python3", "bots/planner.py"]
- executor: ["python3", "bots/executor.py"]
- evaluator: ["python3", "bots/evaluator.py"]
+ planner: ["python3", "roles/planner.py"]
+ executor: ["bash", "roles/executor.sh"]
+ evaluator: ["python3", "roles/evaluator.py"]
diff --git a/roles/evaluator.py b/roles/evaluator.py
new file mode 100755
index 0000000..1219394
--- /dev/null
+++ b/roles/evaluator.py
@@ -0,0 +1,124 @@
+#!/usr/bin/env python3
+"""LLM evaluator role.
+
+Reads evaluator_input.json, calls an LLM to judge accept/reject, writes evaluation.json.
+LLM provider/model are read from config.json in the same run directory.
+Reports cost_usd and tokens_used so the kernel can enforce cost guards.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import re
+import sys
+from pathlib import Path
+
+
+def _call_llm(prompt: str, cfg: dict) -> tuple[str, int, float]:
+ provider = cfg.get("provider", "anthropic")
+ model = cfg.get("model", "claude-sonnet-4-6")
+ api_key_env = cfg.get("api_key_env", "ANTHROPIC_API_KEY")
+
+ if provider == "anthropic":
+ 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}],
+ )
+ tokens = msg.usage.input_tokens + msg.usage.output_tokens
+ return msg.content[0].text, tokens, tokens * 3e-6
+
+ if provider == "openai":
+ 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}],
+ )
+ tokens = resp.usage.total_tokens
+ return resp.choices[0].message.content, tokens, tokens * 3e-6
+
+ raise ValueError(f"unknown llm.provider: {provider!r}")
+
+
+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", {})
+
+ goal = inp.get("goal", {})
+ patch_path = inp.get("patch_path", "")
+ obs_path = inp.get("observation_path", "")
+
+ patch_text = ""
+ if patch_path and Path(patch_path).exists():
+ patch_text = Path(patch_path).read_text(encoding="utf-8")[:3000]
+
+ obs_text = ""
+ if obs_path and Path(obs_path).exists():
+ obs_text = Path(obs_path).read_text(encoding="utf-8")[:1000]
+
+ prompt = f"""You are a code evolution evaluator. Decide whether to ACCEPT or REJECT a candidate change.
+
+Goal: {goal.get("objective", goal.get("name", ""))}
+
+Observation (current state):
+{obs_text or "(none)"}
+
+Patch applied:
+{patch_text or "(no changes)"}
+
+Respond with ONLY a JSON object:
+- "hard_gates_passed": true if the change is safe and relevant, false otherwise
+- "recommendation": "accept" or "reject"
+- "reason": one sentence explaining your decision
+- "metrics": {{}} (optional key/value metrics you can infer)
+"""
+
+ try:
+ text, tokens, cost = _call_llm(prompt, llm_cfg)
+ m = re.search(r"\{.*\}", text, re.DOTALL)
+ if m:
+ result = json.loads(m.group())
+ else:
+ result = {
+ "hard_gates_passed": False,
+ "recommendation": "reject",
+ "reason": f"evaluator could not parse LLM response: {text[:100]}",
+ "metrics": {},
+ }
+ except Exception as exc:
+ result = {
+ "hard_gates_passed": False,
+ "recommendation": "reject",
+ "reason": f"evaluator error: {exc}",
+ "metrics": {},
+ }
+ tokens, cost = 0, 0.0
+
+ result.setdefault("hard_gates_passed", False)
+ result.setdefault("recommendation", "reject")
+ result.setdefault("reason", "")
+ result.setdefault("metrics", {})
+ result["cost_usd"] = cost
+ result["tokens_used"] = tokens
+
+ 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/executor.sh b/roles/executor.sh
new file mode 100755
index 0000000..f6f7e49
--- /dev/null
+++ b/roles/executor.sh
@@ -0,0 +1,69 @@
+#!/usr/bin/env bash
+# Executor role: calls a configurable coding agent to apply the plan.
+#
+# Reads executor_input.json (--input), writes executor_output.json (--output).
+# Coding agent is selected via `coding_agent.tool` in config.json (same run dir).
+# Supported tools: aider | claude-code
+#
+# Usage: executor.sh --input --output --worktree
+set -euo pipefail
+
+INPUT="" OUTPUT="" WORKTREE=""
+while [[ $# -gt 0 ]]; do
+ case "$1" in
+ --input) INPUT="$2"; shift 2 ;;
+ --output) OUTPUT="$2"; shift 2 ;;
+ --worktree) WORKTREE="$2"; shift 2 ;;
+ *) echo "unknown arg: $1" >&2; exit 1 ;;
+ esac
+done
+
+[[ -n "$INPUT" && -n "$OUTPUT" && -n "$WORKTREE" ]] || { echo "missing required args" >&2; exit 1; }
+
+RUN_DIR="$(dirname "$INPUT")"
+
+# Load plan
+PLAN_PATH="$(python3 -c "import json,sys; d=json.load(open('$INPUT')); print(d.get('plan_path',''))" 2>/dev/null || echo "")"
+if [[ -z "$PLAN_PATH" || ! -f "$PLAN_PATH" ]]; then
+ PLAN_PATH="$RUN_DIR/plan.json"
+fi
+SUMMARY="$(python3 -c "import json; d=json.load(open('$PLAN_PATH')); print(d.get('summary','improve the codebase'))" 2>/dev/null || echo "improve the codebase")"
+STEPS="$(python3 -c "import json; d=json.load(open('$PLAN_PATH')); print('\n'.join(d.get('steps',[])) or 'Apply the plan.')" 2>/dev/null || echo "Apply the plan.")"
+
+# Load coding agent tool from config.json
+TOOL="aider"
+CONFIG_PATH="$RUN_DIR/config.json"
+if [[ -f "$CONFIG_PATH" ]]; then
+ TOOL="$(python3 -c "import json; d=json.load(open('$CONFIG_PATH')); print(d.get('coding_agent',{}).get('tool','aider'))" 2>/dev/null || echo "aider")"
+fi
+
+PROMPT="$SUMMARY
+
+Steps:
+$STEPS
+
+Important: only modify files within the allowed paths specified in the plan."
+
+cd "$WORKTREE"
+
+case "$TOOL" in
+ aider)
+ aider --message "$PROMPT" --yes --no-pretty --auto-commits=false 2>&1 || true
+ ;;
+ claude-code)
+ claude -p "$PROMPT" 2>&1 || true
+ ;;
+ *)
+ echo "error: unknown coding_agent.tool: $TOOL" >&2
+ exit 1
+ ;;
+esac
+
+CHANGED=$(git status --porcelain 2>/dev/null | wc -l | tr -d ' ')
+# Use jq to safely encode strings — avoids shell-quoting bugs when LLM output
+# contains quotes, backslashes, or other special characters.
+jq -n \
+ --argjson changed "$CHANGED" \
+ --arg tool "$TOOL" \
+ --arg summary "$SUMMARY" \
+ '{"changed_files": $changed, "tool": $tool, "summary": $summary}' > "$OUTPUT"
diff --git a/roles/planner.py b/roles/planner.py
new file mode 100755
index 0000000..ec2f14f
--- /dev/null
+++ b/roles/planner.py
@@ -0,0 +1,137 @@
+#!/usr/bin/env python3
+"""LLM planner role.
+
+Reads planner_input.json, calls an LLM to produce a plan, writes plan.json.
+LLM provider/model are read from config.json in the same run directory.
+
+Config keys used (under `llm:`):
+ provider: anthropic (default) | openai
+ model: e.g. claude-sonnet-4-6 or gpt-4o
+ api_key_env: name of the env var holding the API key
+"""
+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) -> tuple[str, int, float]:
+ import anthropic # type: ignore
+ client = anthropic.Anthropic(api_key=os.environ[api_key_env])
+ msg = client.messages.create(
+ model=model,
+ max_tokens=1024,
+ messages=[{"role": "user", "content": prompt}],
+ )
+ tokens = msg.usage.input_tokens + msg.usage.output_tokens
+ # Approximate cost — exact pricing varies by model; callers may override.
+ cost = tokens * 3e-6
+ return msg.content[0].text, tokens, cost
+
+
+def _call_openai(prompt: str, model: str, api_key_env: str) -> tuple[str, int, float]:
+ 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}],
+ )
+ tokens = resp.usage.total_tokens
+ cost = tokens * 3e-6
+ return resp.choices[0].message.content, tokens, cost
+
+
+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"))
+
+ # Load LLM config from run-dir config.json (written by governor).
+ 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")
+
+ goal = inp.get("goal", {})
+ allowed_paths = inp.get("allowed_paths", [])
+ history = inp.get("history", [])
+
+ obs_text = ""
+ obs_path = inp.get("observation_path", "")
+ if obs_path and Path(obs_path).exists():
+ obs_text = Path(obs_path).read_text(encoding="utf-8")
+
+ history_text = "\n".join(
+ f"- Run {h['run_id']}: {'ACCEPTED' if h.get('accepted') else 'REJECTED'} — {h.get('summary', '')}"
+ for h in history
+ ) or "(no history yet — this is the first run)"
+
+ prompt = f"""You are a code evolution planner. Produce a concrete plan to make progress toward the goal.
+
+Goal: {goal.get("objective", goal.get("name", ""))}
+
+Current observation:
+{obs_text or "(none)"}
+
+Allowed paths (ONLY modify files under these paths):
+{json.dumps(allowed_paths)}
+
+Previous attempts:
+{history_text}
+
+Respond with ONLY a JSON object containing:
+- "summary": one-line description of the change
+- "steps": list of concrete implementation steps
+- "expected_improvement": what should improve after this change
+- "allowed_paths": paths to be modified (must be a subset of the allowed list above)
+- "abort": false (set true only if you have absolutely no viable approach)
+"""
+
+ if provider == "anthropic":
+ text, tokens, cost = _call_anthropic(prompt, model, api_key_env)
+ elif provider == "openai":
+ text, tokens, cost = _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)
+ plan = None
+ if m:
+ try:
+ plan = json.loads(m.group())
+ except json.JSONDecodeError:
+ pass
+ if plan is None:
+ plan = {
+ "summary": text[:200],
+ "steps": [text],
+ "expected_improvement": "",
+ "allowed_paths": allowed_paths,
+ "abort": False,
+ }
+
+ plan.setdefault("run_id", inp.get("run_id", ""))
+ plan.setdefault("abort", False)
+ plan.setdefault("allowed_paths", allowed_paths)
+ plan["_tokens_used"] = tokens
+ plan["_cost_usd"] = cost
+
+ Path(args.output).write_text(json.dumps(plan, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+
+
+if __name__ == "__main__":
+ main()
diff --git a/tests/test_pr4.py b/tests/test_pr4.py
new file mode 100644
index 0000000..0fa9682
--- /dev/null
+++ b/tests/test_pr4.py
@@ -0,0 +1,262 @@
+"""Tests for PR4 features: cost guard, history injection, --loop flag, new config fields."""
+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_cost_guard_defaults(self):
+ cfg = parse_config({"mission": "x"})
+ self.assertEqual(cfg.hard_stops.max_total_usd, 0.0)
+ self.assertEqual(cfg.hard_stops.max_total_tokens, 0)
+
+ def test_cost_guard_values(self):
+ cfg = parse_config({"mission": "x", "hard_stops": {
+ "max_iterations": 5, "max_consecutive_failures": 2,
+ "max_total_usd": 1.5, "max_total_tokens": 100000,
+ }})
+ self.assertAlmostEqual(cfg.hard_stops.max_total_usd, 1.5)
+ self.assertEqual(cfg.hard_stops.max_total_tokens, 100000)
+
+ def test_llm_defaults(self):
+ cfg = parse_config({"mission": "x"})
+ self.assertEqual(cfg.llm.provider, "anthropic")
+ self.assertEqual(cfg.llm.model, "claude-sonnet-4-6")
+ self.assertEqual(cfg.llm.api_key_env, "ANTHROPIC_API_KEY")
+
+ def test_llm_custom(self):
+ cfg = parse_config({"mission": "x", "llm": {
+ "provider": "openai", "model": "gpt-4o", "api_key_env": "OPENAI_API_KEY",
+ }})
+ self.assertEqual(cfg.llm.provider, "openai")
+ self.assertEqual(cfg.llm.model, "gpt-4o")
+
+ def test_coding_agent_default(self):
+ cfg = parse_config({"mission": "x"})
+ self.assertEqual(cfg.coding_agent.tool, "aider")
+
+ def test_coding_agent_claude_code(self):
+ cfg = parse_config({"mission": "x", "coding_agent": {"tool": "claude-code"}})
+ self.assertEqual(cfg.coding_agent.tool, "claude-code")
+
+ def test_history_defaults(self):
+ cfg = parse_config({"mission": "x"})
+ self.assertEqual(cfg.history.max_entries, 10)
+
+ def test_history_custom(self):
+ cfg = parse_config({"mission": "x", "history": {"max_entries": 5}})
+ self.assertEqual(cfg.history.max_entries, 5)
+
+
+# ---------------------------------------------------------------------------
+# Hard stops — cost guard
+# ---------------------------------------------------------------------------
+
+class TestCostGuard(unittest.TestCase):
+
+ def test_precheck_blocks_on_usd(self):
+ state = hard_stops.HardStopState(total_usd=1.0)
+ allowed, reason = hard_stops.precheck(state, 10, 3, max_total_usd=1.0)
+ self.assertFalse(allowed)
+ self.assertIn("max_total_usd", reason)
+
+ def test_precheck_blocks_on_tokens(self):
+ state = hard_stops.HardStopState(total_tokens=500000)
+ allowed, reason = hard_stops.precheck(state, 10, 3, max_total_tokens=500000)
+ self.assertFalse(allowed)
+ self.assertIn("max_total_tokens", reason)
+
+ def test_precheck_allows_below_limit(self):
+ state = hard_stops.HardStopState(total_usd=0.5, total_tokens=100)
+ allowed, _ = hard_stops.precheck(state, 10, 3, max_total_usd=1.0, max_total_tokens=500000)
+ self.assertTrue(allowed)
+
+ def test_record_outcome_accumulates_cost(self):
+ state = hard_stops.HardStopState()
+ state = hard_stops.record_outcome(
+ state, accepted=True, max_iterations=10, max_consecutive_failures=3,
+ cost_usd=0.05, tokens_used=1000,
+ )
+ self.assertAlmostEqual(state.total_usd, 0.05)
+ self.assertEqual(state.total_tokens, 1000)
+
+ def test_record_outcome_halts_on_usd(self):
+ state = hard_stops.HardStopState(total_usd=0.95)
+ state = hard_stops.record_outcome(
+ state, accepted=True, max_iterations=10, max_consecutive_failures=3,
+ cost_usd=0.10, tokens_used=0, max_total_usd=1.0,
+ )
+ self.assertTrue(state.halted)
+ self.assertIn("max_total_usd", state.halt_reason)
+
+ def test_record_outcome_halts_on_tokens(self):
+ state = hard_stops.HardStopState(total_tokens=490000)
+ state = hard_stops.record_outcome(
+ state, accepted=True, max_iterations=10, max_consecutive_failures=3,
+ tokens_used=20000, max_total_tokens=500000,
+ )
+ self.assertTrue(state.halted)
+ self.assertIn("max_total_tokens", state.halt_reason)
+
+ def test_state_persists_cost_fields(self):
+ with tempfile.TemporaryDirectory() as tmp:
+ ledger = Path(tmp)
+ state = hard_stops.HardStopState(total_usd=0.12, total_tokens=4500)
+ hard_stops.save_state(ledger, state)
+ loaded = hard_stops.load_state(ledger)
+ self.assertAlmostEqual(loaded.total_usd, 0.12)
+ self.assertEqual(loaded.total_tokens, 4500)
+
+
+# ---------------------------------------------------------------------------
+# Governor — history injection
+# ---------------------------------------------------------------------------
+
+class TestHistoryInjection(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, max_entries: int = 10) -> Governor:
+ return Governor(
+ target_repo=self.repo,
+ ledger_dir=self.ledger,
+ planner=_role("planner.py"),
+ executor=_role("executor.py"),
+ evaluator=_role("evaluator_accept.py"),
+ allowed_paths=["src/"],
+ history_max_entries=max_entries,
+ )
+
+ def test_first_run_has_empty_history(self):
+ gov = self._make_governor()
+ gov.run_once({"name": "test", "objective": "test"})
+ planner_input = json.loads(
+ (self.ledger / "runs" / "0001" / "planner_input.json").read_text()
+ )
+ self.assertEqual(planner_input["history"], [])
+
+ def test_second_run_sees_first_run_in_history(self):
+ gov = self._make_governor()
+ gov.run_once({"name": "test", "objective": "test"})
+ gov.run_once({"name": "test", "objective": "test"})
+ planner_input = json.loads(
+ (self.ledger / "runs" / "0002" / "planner_input.json").read_text()
+ )
+ self.assertEqual(len(planner_input["history"]), 1)
+ self.assertEqual(planner_input["history"][0]["run_id"], "0001")
+
+ def test_history_capped_by_max_entries(self):
+ gov = self._make_governor(max_entries=2)
+ for _ in range(4):
+ gov.run_once({"name": "test", "objective": "test"})
+ planner_input = json.loads(
+ (self.ledger / "runs" / "0004" / "planner_input.json").read_text()
+ )
+ self.assertLessEqual(len(planner_input["history"]), 2)
+
+
+# ---------------------------------------------------------------------------
+# CLI — --loop flag
+# ---------------------------------------------------------------------------
+
+class TestLoopFlag(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)
+
+ # Write a minimal config that uses fixture roles
+ self.config_path = self.base / "evolution.yml"
+ # No allowed_paths restriction so fixture executor (writes EVOLUTION_MARKER.txt) is in scope.
+ self.config_path.write_text(f"""
+mission: "test loop"
+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"]
+""")
+
+ def tearDown(self):
+ self._tmp.cleanup()
+
+ def _run_cli(self, *extra_args):
+ from evolution_kernel.cli import main
+ return main([
+ "--config", str(self.config_path),
+ "--repo", str(self.repo),
+ "--ledger", self.ledger,
+ *extra_args,
+ ])
+
+ def test_loop_runs_until_max_iterations(self):
+ rc = self._run_cli("--loop")
+ self.assertEqual(rc, 3) # halted → exit 3
+ # max_iterations=3, so 3 run dirs should exist
+ runs = list((Path(self.ledger) / "runs").iterdir())
+ self.assertEqual(len(runs), 3)
+
+ def test_loop_state_halted_after_completion(self):
+ self._run_cli("--loop")
+ state = hard_stops.load_state(self.ledger)
+ self.assertTrue(state.halted)
+ self.assertIn("max_iterations", state.halt_reason or "")
+
+
+if __name__ == "__main__":
+ unittest.main()