From c1f538395bea1deb41359447128b47845897723d Mon Sep 17 00:00:00 2001 From: Protocol Zero <257158451+Protocol-zero-0@users.noreply.github.com> Date: Thu, 14 May 2026 03:17:07 +0000 Subject: [PATCH] feat: HTTP evidence source for the observer (closes #18) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a third evidence-source type so the planner can see live HTTP state (eval endpoints, health dashboards, third-party scoring APIs) in the same observation bundle that already carries file and shell output. Design - `evolution_kernel/config.py`: `EvidenceSource` gains HTTP-only fields (`url`, `method`, `headers`, `timeout`). `_parse_evidence_sources` validates the new `type: http` shape with friendly errors. - `evolution_kernel/observer.py`: new `_collect_http()` uses stdlib `urllib.request` only (single-dep rule preserved — still only PyYAML). Captures `status`, `body` (64 KiB cap, `truncated` flag), and a sorted list of `(name, value)` headers so the ledger is stable for diffing. Non-2xx responses still record the body so a planner can react to 4xx/5xx instead of silently retrying. URLError / TimeoutError / OSError land in `error` instead of raising. Tests (16 new, all green) - `tests/test_pr7b.py`: - 8 config parsing tests (defaults, headers map, missing/blank URL, bad timeout, non-positive timeout, bad headers shape, unknown type). - 6 `_collect_http` tests against a `http.server.ThreadingHTTPServer` on `127.0.0.1` in a daemon thread: 200 capture, 500-with-body, custom request headers, body truncation flag, connection-refused error, blank URL. - 2 E2E tests: `collect_observation` happy path and `Governor.run_once` writing the HTTP response into the run's `observation.json`. Whole suite: 99 passed (was 83). Co-Authored-By: Claude Opus 4.7 --- README.md | 8 + README.zh.md | 8 + evolution_kernel/config.py | 62 ++++++- evolution_kernel/observer.py | 57 ++++++ tests/test_pr7b.py | 333 +++++++++++++++++++++++++++++++++++ 5 files changed, 466 insertions(+), 2 deletions(-) create mode 100644 tests/test_pr7b.py diff --git a/README.md b/README.md index 5d1ff71..e6fe639 100644 --- a/README.md +++ b/README.md @@ -102,6 +102,8 @@ evolution-kernel --config evolution.yml --repo /path/to/project --ledger /tmp/le ## See it in action +> 📋 **Illustrative scenario.** The numbers below describe what a complete, well-targeted overnight run on the GSM8K case looks like — they are a design narrative, not a checked-in artifact in this repo. For runs anyone can reproduce today, see [`evidence/`](evidence/) and [`examples/demo_target`](examples/demo_target). + ### $34. One night. An 8B model that runs on a MacBook — from 51.8% to 96.2% on elementary math. Zero weight changes. > Qwen3-8B-Instruct is a general-purpose model with no math-specific training. Its weights are frozen throughout. Evolution Kernel evolves only the solver harness — prompt strategies, tools, and sampling logic. After one overnight run, the same model sits 2.8 points behind GPT-5.5. That means every child can have a free, local, always-on, privacy-safe math tutor. @@ -248,6 +250,7 @@ flowchart LR | Goal evaluator — stops when mission is "won" | ✅ | | k-branch parallel exploration (FunSearch / AlphaEvolve style) | ✅ | | Process sandbox via firejail — executor cannot write outside its worktree | ✅ | +| Remote observer — HTTP evidence source for live dashboards / eval endpoints | ✅ | --- @@ -266,6 +269,11 @@ evidence_sources: command: "python3 scripts/run_gsm8k.py --model qwen3-8b-instruct --sample 100 --json" - type: file # file contents go into observation.json path: "metrics.json" + - type: http # GET a live endpoint; status, headers and body recorded + url: "https://evals.example.com/run/latest" + headers: + Accept: application/json + timeout: 10 # seconds (default 10) # Only files under these paths may be changed mutation_scope: diff --git a/README.zh.md b/README.zh.md index d929d3d..bb7ef12 100644 --- a/README.zh.md +++ b/README.zh.md @@ -102,6 +102,8 @@ evolution-kernel --config evolution.yml --repo /path/to/project --ledger /tmp/le ## 看它实际运行 +> 📋 **示意场景。** 下面的数字描述了"一个完整、目标明确的隔夜跑"是什么样——这是 GSM8K 案例的设计叙事，不是仓库里 checked-in 的真实运行记录。今天就能复现的真实跑，请参考 [`evidence/`](evidence/) 和 [`examples/demo_target`](examples/demo_target)。 + ### $34，一晚上，一个能在 MacBook 上跑的 8B 模型——小学数学应用题正确率 96.2%，和 GPT-5.5 基本同档。模型权重一字节未动。 > Qwen3-8B-Instruct 是一个通用模型，没有专门的数学训练。权重全程冻结。Evolution Kernel 只进化 solver harness——提示策略、工具调用和采样逻辑。一个隔夜跑完，同一个模型只落后 GPT-5.5 2.8 个百分点。这意味着每个孩子都能拥有一个免费、本地、随时在线、完全保护隐私的数学辅导老师。 @@ -248,6 +250,7 @@ flowchart LR | 目标评估器——当 mission 完成时自动停止 | ✅ | | k 路并行探索（FunSearch / AlphaEvolve 模式） | ✅ | | 进程级沙箱（firejail）——执行器无法写出 worktree 之外的任何文件 | ✅ | +| 远程观察者——HTTP 证据源，把线上 dashboard / eval endpoint 拉进 observation.json | ✅ | --- @@ -266,6 +269,11 @@ evidence_sources: command: "python3 scripts/run_gsm8k.py --model qwen3-8b-instruct --sample 100 --json" - type: file # 文件内容写入 observation.json path: "metrics.json" + - type: http # GET 一个线上接口；status / headers / body 都进 observation + url: "https://evals.example.com/run/latest" + headers: + Accept: application/json + timeout: 10 # 秒（默认 10） # 只有这些路径下的文件允许被修改 mutation_scope: diff --git a/evolution_kernel/config.py b/evolution_kernel/config.py index 9b30dd3..2c795cf 100644 --- a/evolution_kernel/config.py +++ b/evolution_kernel/config.py @@ -20,6 +20,12 @@ path: "./metrics.json" - type: shell command: "bash ./scripts/status.sh" + - type: http + url: "http://localhost:8000/status" + method: GET # optional, default GET + headers: # optional + Accept: application/json + timeout: 10 # optional seconds, default 10 mutation_scope: allowed_paths: @@ -58,9 +64,14 @@ class ConfigError(ValueError): @dataclass(frozen=True) class EvidenceSource: - type: str # "file" or "shell" + type: str # "file" | "shell" | "http" path: str | None = None command: str | None = None + # HTTP-only fields + url: str | None = None + method: str = "GET" + headers: tuple[tuple[str, str], ...] = () + timeout: float = 10.0 @dataclass(frozen=True) @@ -208,9 +219,56 @@ def _parse_evidence_sources(value: Any) -> Sequence[EvidenceSource]: f"evidence_sources[{index}] type=shell requires a non-empty `command`" ) sources.append(EvidenceSource(type="shell", command=command.strip())) + elif kind == "http": + url = item.get("url") + if not isinstance(url, str) or not url.strip(): + raise ConfigError( + f"evidence_sources[{index}] type=http requires a non-empty `url`" + ) + method_raw = item.get("method", "GET") + if not isinstance(method_raw, str) or not method_raw.strip(): + raise ConfigError( + f"evidence_sources[{index}].method must be a non-empty string" + ) + headers_raw = item.get("headers", {}) + if not isinstance(headers_raw, Mapping): + raise ConfigError( + f"evidence_sources[{index}].headers must be a mapping" + ) + headers: list[tuple[str, str]] = [] + for hk, hv in headers_raw.items(): + if not isinstance(hk, str) or not hk.strip(): + raise ConfigError( + f"evidence_sources[{index}].headers keys must be non-empty strings" + ) + if not isinstance(hv, (str, int, float)): + raise ConfigError( + f"evidence_sources[{index}].headers[{hk!r}] must be str | int | float" + ) + headers.append((hk.strip(), str(hv))) + timeout_raw = item.get("timeout", 10.0) + try: + timeout = float(timeout_raw) + except (TypeError, ValueError): + raise ConfigError( + f"evidence_sources[{index}].timeout must be a number, got {timeout_raw!r}" + ) + if timeout <= 0: + raise ConfigError( + f"evidence_sources[{index}].timeout must be > 0" + ) + sources.append( + EvidenceSource( + type="http", + url=url.strip(), + method=method_raw.strip().upper(), + headers=tuple(headers), + timeout=timeout, + ) + ) else: raise ConfigError( - f"evidence_sources[{index}].type must be 'file' or 'shell', got {kind!r}" + f"evidence_sources[{index}].type must be 'file', 'shell', or 'http', got {kind!r}" ) return sources diff --git a/evolution_kernel/observer.py b/evolution_kernel/observer.py index 1ee7569..7bfbd04 100644 --- a/evolution_kernel/observer.py +++ b/evolution_kernel/observer.py @@ -8,6 +8,8 @@ import json import subprocess +import urllib.error +import urllib.request from pathlib import Path from typing import Any, Mapping, Sequence @@ -15,6 +17,7 @@ DEFAULT_FILE_LIMIT = 64 * 1024 # 64 KiB DEFAULT_SHELL_TIMEOUT = 30 # seconds +DEFAULT_HTTP_LIMIT = 64 * 1024 # 64 KiB body cap def collect_observation( @@ -22,6 +25,7 @@ def collect_observation( cwd: Path | str, file_limit: int = DEFAULT_FILE_LIMIT, shell_timeout: int = DEFAULT_SHELL_TIMEOUT, + http_limit: int = DEFAULT_HTTP_LIMIT, ) -> Mapping[str, Any]: """Run each evidence source and return a structured bundle.""" base = Path(cwd).resolve() @@ -31,6 +35,8 @@ def collect_observation( collected.append(_collect_file(source.path or "", base, file_limit)) elif source.type == "shell": collected.append(_collect_shell(source.command or "", base, shell_timeout)) + elif source.type == "http": + collected.append(_collect_http(source, http_limit)) else: collected.append({"type": source.type, "error": "unknown source type"}) return {"cwd": str(base), "sources": collected} @@ -73,6 +79,57 @@ def _collect_file(rel_path: str, cwd: Path, limit: int) -> dict[str, Any]: return record +def _collect_http(source: EvidenceSource, limit: int) -> dict[str, Any]: + """Fetch one HTTP endpoint into a structured record. + + The observer stays a thin collector: it captures status, headers, and a + byte-capped body, plus an ``error`` string on failure. Retries, auth + flows, and content parsing belong upstream (planner / role scripts). + """ + url = source.url or "" + method = (source.method or "GET").upper() + record: dict[str, Any] = {"type": "http", "url": url, "method": method} + if not url.strip(): + record["error"] = "empty url" + return record + request = urllib.request.Request(url, method=method) + for header_name, header_value in source.headers: + request.add_header(header_name, header_value) + try: + with urllib.request.urlopen(request, timeout=source.timeout) as response: + status = int(getattr(response, "status", response.getcode() or 0)) + headers_obj = response.headers + body = response.read(limit + 1) + except urllib.error.HTTPError as exc: + # Non-2xx responses still carry a body — record it like a success + # except for the status code so the planner can react to 4xx/5xx. + try: + body = exc.read(limit + 1) + except Exception: + body = b"" + headers_obj = exc.headers + status = int(exc.code) + except urllib.error.URLError as exc: + record["error"] = f"URLError: {exc.reason!r}" + return record + except (TimeoutError, OSError) as exc: + record["error"] = f"{type(exc).__name__}: {exc}" + return record + + if len(body) > limit: + record["truncated"] = True + body = body[:limit] + record["status"] = status + record["bytes"] = len(body) + record["body"] = body.decode("utf-8", errors="replace") + # Sort headers so the observation bundle is stable for ledger diffs. + record["headers"] = sorted( + ((str(k), str(v)) for k, v in headers_obj.items()), + key=lambda kv: (kv[0].lower(), kv[1]), + ) + return record + + def _collect_shell(command: str, cwd: Path, timeout: int) -> dict[str, Any]: record: dict[str, Any] = {"type": "shell", "command": command} if not command.strip(): diff --git a/tests/test_pr7b.py b/tests/test_pr7b.py new file mode 100644 index 0000000..86b296b --- /dev/null +++ b/tests/test_pr7b.py @@ -0,0 +1,333 @@ +"""Tests for Issue #18 / PR7b: HTTP evidence source in the observer. + +Layers exercised: + +1. Config parsing — ``type: http`` shape, defaults, validation errors. +2. ``observer._collect_http`` — directly invoked against a local + ``http.server.ThreadingHTTPServer`` running on an ephemeral port in a + daemon thread. Covers happy path, 5xx with body, and connection-refused. +3. End-to-end ``Governor.run_once`` with a single ``type: http`` evidence + source — asserts the recorded observation.json reflects the HTTP response. + +The new source uses ``urllib.request`` from the stdlib, so no new third-party +dependency is added to the kernel. +""" +from __future__ import annotations + +import json +import socket +import subprocess +import sys +import tempfile +import threading +import unittest +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path + +from evolution_kernel.config import ConfigError, EvidenceSource, parse_config +from evolution_kernel.governor import Governor, RoleCommand +from evolution_kernel.observer import _collect_http, collect_observation + + +ROOT = Path(__file__).resolve().parents[1] +FIXTURES = ROOT / "tests" / "fixtures" + + +def _role(name: str) -> RoleCommand: + return RoleCommand([sys.executable, str(FIXTURES / name)]) + + +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", encoding="utf-8") + _git(["add", "-A"], repo) + _git(["commit", "-m", "initial"], repo) + + +# --------------------------------------------------------------------------- +# Local HTTP test server +# --------------------------------------------------------------------------- + + +class _EchoHandler(BaseHTTPRequestHandler): + """A handler that lets each test choose status + body via path. + + /ok → 200 with a known body + /500 → 500 with a body + /headers → 200 echoing inbound headers, so we can verify header passing + anything else → 404 with empty body + """ + + server_version = "EvolutionKernelTestServer/1.0" + + def log_message(self, format, *args): # silence test noise + return + + def do_GET(self): # noqa: N802 — stdlib mandates this name + if self.path == "/ok": + body = b'{"status":"ok","score":0.42}' + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("X-Test-Marker", "evolution-kernel") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + elif self.path == "/500": + body = b"internal failure body" + self.send_response(500) + self.send_header("Content-Type", "text/plain") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + elif self.path == "/headers": + received = self.headers.get("X-Auth", "") + body = f"X-Auth={received}".encode("utf-8") + self.send_response(200) + self.send_header("Content-Type", "text/plain") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + else: + self.send_response(404) + self.send_header("Content-Length", "0") + self.end_headers() + + +class _ServerHandle: + def __init__(self): + self.server = ThreadingHTTPServer(("127.0.0.1", 0), _EchoHandler) + self.port = self.server.server_address[1] + self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) + self.thread.start() + + @property + def base_url(self) -> str: + return f"http://127.0.0.1:{self.port}" + + def close(self): + self.server.shutdown() + self.server.server_close() + + +# --------------------------------------------------------------------------- +# Config parsing +# --------------------------------------------------------------------------- + + +class TestHttpSourceConfigParsing(unittest.TestCase): + + def test_minimal_http_source_with_defaults(self): + cfg = parse_config({ + "mission": "x", + "evidence_sources": [ + {"type": "http", "url": "http://localhost:8000/status"}, + ], + }) + (src,) = cfg.evidence_sources + self.assertEqual(src.type, "http") + self.assertEqual(src.url, "http://localhost:8000/status") + self.assertEqual(src.method, "GET") + self.assertEqual(src.headers, ()) + self.assertEqual(src.timeout, 10.0) + + def test_full_http_source_parsed(self): + cfg = parse_config({ + "mission": "x", + "evidence_sources": [{ + "type": "http", + "url": "http://localhost:8000/eval", + "method": "post", + "headers": {"Accept": "application/json", "X-Run": 42}, + "timeout": 5, + }], + }) + (src,) = cfg.evidence_sources + self.assertEqual(src.method, "POST") # uppercased + self.assertEqual(dict(src.headers), {"Accept": "application/json", "X-Run": "42"}) + self.assertEqual(src.timeout, 5.0) + + def test_missing_url_rejected(self): + with self.assertRaises(ConfigError): + parse_config({ + "mission": "x", + "evidence_sources": [{"type": "http"}], + }) + + def test_blank_url_rejected(self): + with self.assertRaises(ConfigError): + parse_config({ + "mission": "x", + "evidence_sources": [{"type": "http", "url": " "}], + }) + + def test_bad_timeout_rejected(self): + with self.assertRaises(ConfigError): + parse_config({ + "mission": "x", + "evidence_sources": [{"type": "http", "url": "http://x", "timeout": "soon"}], + }) + + def test_non_positive_timeout_rejected(self): + with self.assertRaises(ConfigError): + parse_config({ + "mission": "x", + "evidence_sources": [{"type": "http", "url": "http://x", "timeout": 0}], + }) + + def test_bad_headers_shape_rejected(self): + with self.assertRaises(ConfigError): + parse_config({ + "mission": "x", + "evidence_sources": [{"type": "http", "url": "http://x", "headers": [1, 2]}], + }) + + def test_unknown_type_rejected(self): + with self.assertRaises(ConfigError): + parse_config({ + "mission": "x", + "evidence_sources": [{"type": "telnet", "url": "x"}], + }) + + +# --------------------------------------------------------------------------- +# _collect_http unit tests +# --------------------------------------------------------------------------- + + +class TestCollectHttp(unittest.TestCase): + + @classmethod + def setUpClass(cls): + cls.server = _ServerHandle() + + @classmethod + def tearDownClass(cls): + cls.server.close() + + def test_200_captures_status_body_headers(self): + rec = _collect_http( + EvidenceSource(type="http", url=f"{self.server.base_url}/ok"), + limit=64 * 1024, + ) + self.assertEqual(rec["status"], 200) + self.assertEqual(rec["method"], "GET") + self.assertIn('"score":0.42', rec["body"]) + # Headers list contains the marker we set on the server side. + kv = {k.lower(): v for k, v in rec["headers"]} + self.assertEqual(kv.get("x-test-marker"), "evolution-kernel") + self.assertNotIn("error", rec) + + def test_500_still_records_body(self): + rec = _collect_http( + EvidenceSource(type="http", url=f"{self.server.base_url}/500"), + limit=64 * 1024, + ) + self.assertEqual(rec["status"], 500) + self.assertEqual(rec["body"], "internal failure body") + self.assertNotIn("error", rec) + + def test_headers_are_sent(self): + rec = _collect_http( + EvidenceSource( + type="http", + url=f"{self.server.base_url}/headers", + headers=(("X-Auth", "token-abc"),), + ), + limit=64 * 1024, + ) + self.assertEqual(rec["status"], 200) + self.assertEqual(rec["body"], "X-Auth=token-abc") + + def test_truncation_marked_when_body_exceeds_limit(self): + rec = _collect_http( + EvidenceSource(type="http", url=f"{self.server.base_url}/ok"), + limit=5, + ) + self.assertTrue(rec.get("truncated")) + self.assertEqual(rec["bytes"], 5) + + def test_connection_refused_records_error(self): + # Bind a socket to grab a free port, then close it so the address is unused. + s = socket.socket() + s.bind(("127.0.0.1", 0)) + free_port = s.getsockname()[1] + s.close() + rec = _collect_http( + EvidenceSource(type="http", url=f"http://127.0.0.1:{free_port}/", timeout=2.0), + limit=1024, + ) + self.assertIn("error", rec) + # No body / status when the connection never opened. + self.assertNotIn("status", rec) + + def test_blank_url_records_error(self): + rec = _collect_http(EvidenceSource(type="http", url=""), limit=1024) + self.assertEqual(rec.get("error"), "empty url") + + +# --------------------------------------------------------------------------- +# End-to-end through collect_observation + Governor.run_once +# --------------------------------------------------------------------------- + + +class TestHttpObservationE2E(unittest.TestCase): + + @classmethod + def setUpClass(cls): + cls.server = _ServerHandle() + + @classmethod + def tearDownClass(cls): + cls.server.close() + + def test_collect_observation_includes_http_response(self): + obs = collect_observation( + (EvidenceSource(type="http", url=f"{self.server.base_url}/ok"),), + cwd=Path("/tmp"), + ) + self.assertEqual(len(obs["sources"]), 1) + src = obs["sources"][0] + self.assertEqual(src["status"], 200) + self.assertIn('"score":0.42', src["body"]) + + def test_governor_run_once_writes_http_into_observation_json(self): + with tempfile.TemporaryDirectory() as td: + base = Path(td) + repo = base / "repo" + ledger = base / "ledger" + _bootstrap_repo(repo) + + governor = Governor( + target_repo=repo, + ledger_dir=ledger, + planner=_role("planner.py"), + executor=_role("executor.py"), + evaluator=_role("evaluator_accept.py"), + evidence_sources=( + EvidenceSource(type="http", url=f"{self.server.base_url}/ok"), + ), + ) + result = governor.run_once({"name": "http-e2e"}, run_id="0001") + + obs_path = result.run_dir / "observation.json" + data = json.loads(obs_path.read_text(encoding="utf-8")) + self.assertEqual(len(data["sources"]), 1) + src = data["sources"][0] + self.assertEqual(src["type"], "http") + self.assertEqual(src["status"], 200) + self.assertEqual(src["url"], f"{self.server.base_url}/ok") + self.assertIn("score", src["body"]) + + +if __name__ == "__main__": + unittest.main()