From f3960a460511eb488e181968d6d5001ebc33d16a Mon Sep 17 00:00:00 2001 From: Cognis Digital <215970675+cognis-digital@users.noreply.github.com> Date: Fri, 3 Jul 2026 21:09:03 -0400 Subject: [PATCH] Multiply capabilities: stop conditions, retry, observers, restarts, checkpoint, strategies, CLI (v0.2) An additive, backward-compatible expansion of the refinement engine. Every new piece plugs into an existing seam (stop= / retry= / on_iteration=) or wraps Engine.run from the outside, so all pre-0.2 callers keep working unchanged. - Pluggable, composable stop conditions (stopping.py): MaxSeconds, TargetScore, NoImprovement (abs + relative delta), ScoreThreshold divergence guard, Predicate; combine with `|`. New Status.TIMEOUT / Status.ABORTED. - Retry/backoff for flaky stages (policy.py): RetryPolicy with bounded exponential backoff, optional jitter, injectable sleep; NO_RETRY default. - Observer ecosystem (observers.py): multi() fan-out, JsonlWriter durable JSONL audit trace, MetricsCollector live stats, ProgressPrinter. - Keep-best-of-N / random restarts (search.py): best_of + parallel_best_of (thread pool for I/O-bound stages), BestOf bundle. - Checkpoint/resume (checkpoint.py): atomic on_iteration observer + resume_state. - Strategy library (strategies.py): hill_climb, simulated_anneal, grid_step, generic_hill_climb revise factories. - CLI (cli.py, __main__.py): `python -m cyclework` sqrt/optimize/demo/version with --target/--max-seconds/--jsonl; console_scripts entry point. - Richer trace: Result.metrics() and Result.to_dict(). - 107 tests (was 24); 6 new narrated demos (11 total); README/ARCHITECTURE/DEMOS updated; version 0.2.0. --- README.md | 121 ++++++++++++++++- cyclework/__init__.py | 40 +++++- cyclework/__main__.py | 5 + cyclework/checkpoint.py | 119 +++++++++++++++++ cyclework/cli.py | 228 +++++++++++++++++++++++++++++++++ cyclework/engine.py | 51 +++++++- cyclework/observers.py | 167 ++++++++++++++++++++++++ cyclework/policy.py | 90 +++++++++++++ cyclework/search.py | 120 +++++++++++++++++ cyclework/stopping.py | 214 +++++++++++++++++++++++++++++++ cyclework/strategies.py | 139 ++++++++++++++++++++ cyclework/trace.py | 55 +++++++- demos/06_wallclock_deadline.py | 36 ++++++ demos/07_retry_backoff.py | 49 +++++++ demos/08_jsonl_audit_trace.py | 49 +++++++ demos/09_strategy_library.py | 47 +++++++ demos/10_best_of_restarts.py | 46 +++++++ demos/11_checkpoint_resume.py | 54 ++++++++ docs/ARCHITECTURE.md | 34 ++++- docs/DEMOS.md | 19 ++- pyproject.toml | 5 +- tests/test_checkpoint.py | 94 ++++++++++++++ tests/test_cli.py | 73 +++++++++++ tests/test_observers.py | 112 ++++++++++++++++ tests/test_policy.py | 110 ++++++++++++++++ tests/test_search.py | 106 +++++++++++++++ tests/test_stopping.py | 180 ++++++++++++++++++++++++++ tests/test_strategies.py | 114 +++++++++++++++++ tests/test_trace_extras.py | 64 +++++++++ 29 files changed, 2516 insertions(+), 25 deletions(-) create mode 100644 cyclework/__main__.py create mode 100644 cyclework/checkpoint.py create mode 100644 cyclework/cli.py create mode 100644 cyclework/observers.py create mode 100644 cyclework/policy.py create mode 100644 cyclework/search.py create mode 100644 cyclework/stopping.py create mode 100644 cyclework/strategies.py create mode 100644 demos/06_wallclock_deadline.py create mode 100644 demos/07_retry_backoff.py create mode 100644 demos/08_jsonl_audit_trace.py create mode 100644 demos/09_strategy_library.py create mode 100644 demos/10_best_of_restarts.py create mode 100644 demos/11_checkpoint_resume.py create mode 100644 tests/test_checkpoint.py create mode 100644 tests/test_cli.py create mode 100644 tests/test_observers.py create mode 100644 tests/test_policy.py create mode 100644 tests/test_search.py create mode 100644 tests/test_stopping.py create mode 100644 tests/test_strategies.py create mode 100644 tests/test_trace_extras.py diff --git a/README.md b/README.md index 195c50d..efd80cb 100644 --- a/README.md +++ b/README.md @@ -101,14 +101,15 @@ model, stopping conditions, and why the loop is a first-class object). | **`seed(initial) -> state`** *(optional)* | A one-time preprocess of the starting candidate. | | **`Engine.run(initial) -> Result`** | Runs the loop to completion. | | **`Engine.cycle(initial)`** | Generator form: yields each `Iteration` as it happens, for streaming/observability. | -| **`Result`** | `status` (`solved` / `plateau` / `exhausted` / `error`), final `state`, `history`, `best`, and `error`. | +| **`Result`** | `status` (`solved` / `plateau` / `exhausted` / `error` / `timeout` / `aborted`), final `state`, `history`, `best`, and `error`. `.metrics()` gives score aggregates; `.to_dict()` a JSON snapshot. | ### Stopping conditions -- **Solved** — a verdict comes back `ok=True`. +- **Solved** — a verdict comes back `ok=True` (or a `TargetScore` / `Predicate` stop condition fires). - **Plateau** — with `patience=N` set, the score fails to improve by at least `min_delta` for `N` consecutive checks. (Higher-is-better; negate a cost to minimize it.) - **Exhausted** — `max_iterations` reached without solving. -- **Error** — a `check`/`revise` raised; the run aborts cleanly with the trace so far. +- **Error** — a `check`/`revise` raised; the run aborts cleanly with the trace so far. A `RetryPolicy` can absorb transient raises first. +- **Timeout / Aborted** — a pluggable `StopCondition` (wall-clock deadline, divergence guard, custom rule) halted the run. See [New in 0.2](#new-in-02--the-loop-fully-equipped). ### Observability @@ -118,14 +119,113 @@ Pass `on_iteration=callback` to receive every `Iteration` as it's recorded — w Engine(check, revise, on_iteration=lambda it: print(it.index, it.score)) ``` +## New in 0.2 — the loop, fully equipped + +The core loop is unchanged and every existing caller keeps working. 0.2 adds +opt-in, composable pieces around it — each is pure standard library and slots in +without touching the engine's contract. + +### Pluggable stop conditions + +Stopping is now composable and domain-driven. A `StopCondition` is any +`stop(iteration, history) -> Optional[Status]`; the ready-made ones cover the +common asks and combine with `|` (first to fire wins). They are consulted *after* +the built-in `ok` check, so `SOLVED` always takes precedence. + +```python +from cyclework import Engine, MaxSeconds, TargetScore, NoImprovement + +stop = MaxSeconds(30) | TargetScore(0.99) | NoImprovement(patience=5) +Engine(check, revise, stop=stop) # halts on whichever trips first +``` + +Also included: `ScoreThreshold` (divergence/bail-out guard) and `Predicate` +(stop when a rule over the candidate is true). + +### Retry with backoff for flaky stages + +Wrap a `check`/`revise` that calls something transient (a model endpoint, a +compiler, a network probe) so one hiccup doesn't abort the run. + +```python +from cyclework import Engine, RetryPolicy + +Engine(check, revise, retry=RetryPolicy(max_attempts=3, base_delay=0.1, jitter=0.05)) +``` + +### Observers: metrics, JSONL audit trace, progress + +`on_iteration` grows into a small ecosystem. `multi(...)` fans one run out to +several sinks; `JsonlWriter` appends one durable JSON line per iteration; +`MetricsCollector` exposes live running stats; `ProgressPrinter` prints a tidy +line per step. + +```python +from cyclework import Engine, multi, JsonlWriter, MetricsCollector + +metrics = MetricsCollector() +with JsonlWriter.to_path("run.jsonl") as trace: + Engine(check, revise, on_iteration=multi(trace, metrics)).run(seed) +print(metrics.snapshot()) +``` + +### Keep-best-of-N and random restarts + +Escape poor local optima by running the loop several times and keeping the best +outcome — serially or concurrently on a thread pool for I/O-bound stages. + +```python +from cyclework import best_of, parallel_best_of + +bundle = best_of(lambda: Engine(check, revise), seeds=[0.0, 2.0, 5.0]) +bundle = parallel_best_of(lambda: Engine(check, revise), seeds=range(8)) +print(bundle.best.summary(), bundle.best_index) +``` + +### Checkpoint / resume + +Persist progress atomically as an observer so a killed process can pick up where +it left off instead of from the seed. + +```python +from cyclework import Engine, Checkpoint + +cp = Checkpoint("run.ckpt", every=5) +start = Checkpoint.resume_state("run.ckpt", default=seed) # resume or seed +Engine(check, revise, on_iteration=cp).run(start) +``` + +### Strategy library + +Ready-made, domain-independent `revise` factories so you don't have to hand-roll +search: `hill_climb`, `simulated_anneal`, `grid_step`, and `generic_hill_climb` +(neighbor search over any state type). + +```python +from cyclework import Engine, strategies + +Engine(check, strategies.simulated_anneal(step=0.3, seed=0)).run(0.0) +``` + +### Command-line front-end + +```bash +python -m cyclework sqrt 2 # Newton's-method sqrt as a loop +python -m cyclework optimize --strategy anneal --restarts 5 +python -m cyclework demo --jsonl trace.jsonl # writes a JSONL audit trace +``` + +Global options (`--max-iter`, `--target`, `--max-seconds`, `--jsonl`, `--quiet`) +work before or after the subcommand. + ## Demos -Five runnable, narrated scenarios in [`demos/`](demos/), each for a different +Eleven runnable, narrated scenarios in [`demos/`](demos/), each for a different audience — all driving the real API, no network, exit 0. Full descriptions in [`docs/DEMOS.md`](docs/DEMOS.md). ```bash -python demos/run_all.py # all five, end to end +python demos/run_all.py # all of them, end to end python demos/02_numeric_solvers.py # or just one ``` @@ -136,6 +236,12 @@ python demos/02_numeric_solvers.py # or just one | 3 | [`03_plateau_and_budget.py`](demos/03_plateau_and_budget.py) | Production loop owners | Honest stopping: SOLVED / PLATEAU / EXHAUSTED / ERROR, always keeping the best candidate. | | 4 | [`04_feedback_refiner.py`](demos/04_feedback_refiner.py) | Pipeline engineers | Feedback-driven revision: the verdict names the fix, revise applies exactly it. | | 5 | [`05_streaming_observability.py`](demos/05_streaming_observability.py) | Observability engineers | Watching a loop live via `on_iteration` and the `Engine.cycle()` generator. | +| 6 | [`06_wallclock_deadline.py`](demos/06_wallclock_deadline.py) | Production loop owners | Composable stop conditions: a wall-clock deadline OR'd with a target score. | +| 7 | [`07_retry_backoff.py`](demos/07_retry_backoff.py) | Reliability engineers | A `RetryPolicy` absorbing a flaky stage's transient failures with backoff. | +| 8 | [`08_jsonl_audit_trace.py`](demos/08_jsonl_audit_trace.py) | Observability / compliance | A durable JSONL audit trace + live `MetricsCollector`, fanned out via `multi`. | +| 9 | [`09_strategy_library.py`](demos/09_strategy_library.py) | Optimization users | Swapping in `hill_climb` / `simulated_anneal` / `grid_step` revise strategies. | +| 10 | [`10_best_of_restarts.py`](demos/10_best_of_restarts.py) | Optimization users | Random restarts with `best_of` escaping a local optimum. | +| 11 | [`11_checkpoint_resume.py`](demos/11_checkpoint_resume.py) | Long-run / fault-tolerance | Checkpointing progress and resuming after a simulated crash. | On a cp1252 (Windows) console, prefix with `PYTHONUTF8=1`. @@ -143,11 +249,12 @@ On a cp1252 (Windows) console, prefix with `PYTHONUTF8=1`. ```bash pip install -e ".[dev]" -pytest -q # 24 tests (engine, examples, and demo smoke tests) +pytest -q # 107 tests (engine, stop conditions, retry, observers, + # search/restarts, checkpoint, strategies, CLI, and demos) ``` ## License COCL (Cognis Open Collaboration License). © Cognis Digital. -> Status: v0.1 — runnable and tested. Roadmap: async stages, parallel candidate fan-out (keep-best-of-N per cycle), and built-in budget accounting (token/time cost per iteration). +> Status: v0.2 — runnable and tested (107 tests, zero dependencies). 0.2 added pluggable/composable stop conditions, retry+backoff, an observer ecosystem (JSONL audit trace, live metrics), keep-best-of-N / parallel restarts, checkpoint-resume, a strategy library, and a CLI — all additive and backward-compatible. Roadmap: async stages and built-in budget accounting (token/time cost per iteration). diff --git a/cyclework/__init__.py b/cyclework/__init__.py index 6925034..a3afebd 100644 --- a/cyclework/__init__.py +++ b/cyclework/__init__.py @@ -18,6 +18,42 @@ from .verdict import Verdict from .trace import Iteration, Result, Status from .engine import Engine +from .stopping import ( + StopCondition, + MaxSeconds, + TargetScore, + NoImprovement, + ScoreThreshold, + Predicate, + as_stop_condition, +) +from .policy import RetryPolicy, NO_RETRY +from .observers import ( + multi, + JsonlWriter, + MetricsCollector, + ProgressPrinter, +) +from .checkpoint import Checkpoint, CheckpointData +from .search import best_of, parallel_best_of, BestOf +from . import strategies -__version__ = "0.1.0" -__all__ = ["Engine", "Verdict", "Result", "Iteration", "Status", "__version__"] +__version__ = "0.2.0" +__all__ = [ + # core + "Engine", "Verdict", "Result", "Iteration", "Status", + # stop conditions + "StopCondition", "MaxSeconds", "TargetScore", "NoImprovement", + "ScoreThreshold", "Predicate", "as_stop_condition", + # retry policy + "RetryPolicy", "NO_RETRY", + # observers + "multi", "JsonlWriter", "MetricsCollector", "ProgressPrinter", + # checkpointing + "Checkpoint", "CheckpointData", + # restart / parallel search + "best_of", "parallel_best_of", "BestOf", + # strategy library + "strategies", + "__version__", +] diff --git a/cyclework/__main__.py b/cyclework/__main__.py new file mode 100644 index 0000000..afc2ffd --- /dev/null +++ b/cyclework/__main__.py @@ -0,0 +1,5 @@ +"""Enable ``python -m cyclework``.""" +from .cli import main + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/cyclework/checkpoint.py b/cyclework/checkpoint.py new file mode 100644 index 0000000..c51a8e3 --- /dev/null +++ b/cyclework/checkpoint.py @@ -0,0 +1,119 @@ +"""Checkpoint / resume. + +A long or fragile refinement run should not have to start over if the process +dies. A :class:`Checkpoint` periodically persists the *current candidate state* +(and the best seen) to disk, so a later invocation can resume from where it left +off rather than from the original seed. + +Design: checkpointing is an *observer* (``callable(Iteration)``) — it slots into +``Engine(on_iteration=...)`` and needs no engine changes. Because a candidate +can be anything, you supply ``dump``/``load`` codecs (default: JSON for +JSON-native states). To resume, load the checkpoint and pass its ``state`` as +the engine's ``initial``. +""" + +from __future__ import annotations + +import json +import os +import tempfile +from dataclasses import dataclass +from typing import Any, Callable, Optional + +from .trace import Iteration + + +@dataclass +class CheckpointData: + index: int + state: Any + best_score: Optional[float] + best_state: Any + + +class Checkpoint: + """Persist run progress every ``every`` iterations (and on the final/ok one). + + Parameters + ---------- + path : where to write the checkpoint (atomically, via a temp file + rename). + every : write a checkpoint every N iterations (default 1). + dump/load : codecs mapping a payload dict <-> bytes. Defaults use JSON, which + works when the state is JSON-native; pass ``pickle`` codecs for arbitrary + objects. + """ + + def __init__( + self, + path: str, + every: int = 1, + dump: Optional[Callable[[dict], bytes]] = None, + load: Optional[Callable[[bytes], dict]] = None, + ): + if every < 1: + raise ValueError("every must be >= 1") + self.path = path + self.every = every + self._dump = dump or _json_dump + self._load = load or _json_load + self._best_score: Optional[float] = None + self._best_state: Any = None + self.writes = 0 + + def __call__(self, it: Iteration) -> None: + s = it.score + if s is not None and (self._best_score is None or s > self._best_score): + self._best_score = s + self._best_state = it.state + if it.verdict.ok or it.index % self.every == 0: + self._write(it) + + def _write(self, it: Iteration) -> None: + payload = { + "index": it.index, + "state": it.state, + "best_score": self._best_score, + "best_state": self._best_state, + } + data = self._dump(payload) + # atomic replace so a crash mid-write never corrupts the checkpoint + d = os.path.dirname(os.path.abspath(self.path)) or "." + fd, tmp = tempfile.mkstemp(dir=d, suffix=".tmp") + try: + with os.fdopen(fd, "wb") as f: + f.write(data) + os.replace(tmp, self.path) + finally: + if os.path.exists(tmp): + os.remove(tmp) + self.writes += 1 + + def load(self) -> Optional[CheckpointData]: + """Read the checkpoint back, or ``None`` if none exists yet.""" + if not os.path.exists(self.path): + return None + with open(self.path, "rb") as f: + payload = self._load(f.read()) + return CheckpointData( + index=payload["index"], + state=payload["state"], + best_score=payload.get("best_score"), + best_state=payload.get("best_state"), + ) + + @staticmethod + def resume_state(path: str, default: Any, + load: Optional[Callable[[bytes], dict]] = None) -> Any: + """Convenience: return the checkpointed candidate state at ``path`` to + resume from, or ``default`` (the original seed) if there is none.""" + cp = Checkpoint(path, load=load) + data = cp.load() + return data.state if data is not None else default + + +def _json_dump(payload: dict) -> bytes: + return json.dumps(payload, ensure_ascii=False).encode("utf-8") + + +def _json_load(data: bytes) -> dict: + return json.loads(data.decode("utf-8")) diff --git a/cyclework/cli.py b/cyclework/cli.py new file mode 100644 index 0000000..ddbc863 --- /dev/null +++ b/cyclework/cli.py @@ -0,0 +1,228 @@ +"""A small command-line front-end: ``python -m cyclework ``. + +The library is domain-agnostic, so the CLI exposes the *engine's* behavior on a +couple of built-in, self-contained problems rather than pretending to be a +general solver. It exists to (a) let you see a run and its trace without writing +code and (b) exercise the new features (stop conditions, strategies, JSONL trace +export, best-of restarts) from a shell. + +Commands +-------- +sqrt N Newton's-method square root as a refinement loop. +optimize Maximize a bumpy 1-D function via a chosen strategy + (hill-climb | anneal | grid), with optional restarts. +demo Run a representative loop and print its trace + metrics. +version Print the library version. + +Global options: --max-iter, --target, --max-seconds, --jsonl PATH, --quiet. +Always exits 0 on a well-formed invocation; exits 2 on a usage error. +""" + +from __future__ import annotations + +import argparse +import json +import math +import sys +from typing import List, Optional + +from . import __version__, strategies +from .engine import Engine +from .observers import JsonlWriter, MetricsCollector, ProgressPrinter, multi +from .search import best_of +from .stopping import MaxSeconds, TargetScore +from .verdict import Verdict + + +def _build_stop(args): + conds = [] + if args.target is not None: + conds.append(TargetScore(args.target)) + if args.max_seconds is not None: + conds.append(MaxSeconds(args.max_seconds)) + if not conds: + return None + stop = conds[0] + for c in conds[1:]: + stop = stop | c + return stop + + +def _observers(args): + obs = [] + writer = None + if args.jsonl: + writer = JsonlWriter.to_path(args.jsonl, include_state=True) + obs.append(writer) + if not args.quiet: + obs.append(ProgressPrinter(every=max(1, args.print_every))) + metrics = MetricsCollector() + obs.append(metrics) + return multi(*obs), metrics, writer + + +def cmd_sqrt(args) -> int: + n = args.n + if n < 0: + print("n must be >= 0", file=sys.stderr) + return 2 + tol = args.tol + + def check(x: float) -> Verdict: + err = abs(x * x - n) + return Verdict(ok=err <= tol, score=-err, detail={"error": err}) + + def revise(x: float, _v: Verdict) -> float: + return (x + n / x) / 2 if x != 0 else 1.0 + + on_it, metrics, writer = _observers(args) + eng = Engine(check, revise, max_iterations=args.max_iter, + stop=_build_stop(args), on_iteration=on_it) + res = eng.run(max(n, 1.0)) + if writer: + writer.close() + _report(res.state, res, metrics, args) + return 0 + + +def _bumpy(x: float) -> float: + """A multi-modal objective: global max near x=2, with local bumps.""" + return math.sin(3 * x) - 0.1 * (x - 2) ** 2 + + +def cmd_optimize(args) -> int: + target = args.target + + def check(x: float) -> Verdict: + s = _bumpy(x) + return Verdict(ok=False, score=s, detail={"x": x}) + + if args.strategy == "hill": + revise = strategies.hill_climb(step=args.step) + elif args.strategy == "anneal": + revise = strategies.simulated_anneal(step=args.step, seed=args.seed) + elif args.strategy == "grid": + lo, hi = args.range + vals = [lo + (hi - lo) * i / (args.max_iter - 1) for i in range(args.max_iter)] + revise = strategies.grid_step(vals) + else: # pragma: no cover - argparse restricts choices + print(f"unknown strategy {args.strategy}", file=sys.stderr) + return 2 + + on_it, metrics, writer = _observers(args) + stop = _build_stop(args) + + def factory() -> Engine: + # fresh engine per restart so stateful strategies reset + if args.strategy == "hill": + rv = strategies.hill_climb(step=args.step) + elif args.strategy == "anneal": + rv = strategies.simulated_anneal(step=args.step, seed=args.seed) + else: + lo, hi = args.range + vals = [lo + (hi - lo) * i / (args.max_iter - 1) for i in range(args.max_iter)] + rv = strategies.grid_step(vals) + return Engine(check, rv, max_iterations=args.max_iter, stop=stop, + on_iteration=on_it if args.restarts <= 1 else None) + + if args.restarts <= 1: + res = factory().run(args.start) + if writer: + writer.close() + _report(res.best.state if res.best else res.state, res, metrics, args) + return 0 + + lo, hi = args.range + seeds = [lo + (hi - lo) * i / max(1, args.restarts - 1) for i in range(args.restarts)] + bundle = best_of(factory(), seeds, stop_when_solved=False) + if writer: + writer.close() + if not args.quiet: + print(json.dumps(bundle.summary(), indent=2)) + best = bundle.best + xbest = best.best.state if best.best else best.state + print(f"best x={xbest:.6f} f(x)={_bumpy(xbest):.6f} over {bundle.summary()['runs']} restarts") + return 0 + + +def cmd_demo(args) -> int: + """A guided run showing off stop conditions + metrics on the bumpy fn.""" + def check(x: float) -> Verdict: + return Verdict(ok=False, score=_bumpy(x), detail={"x": x}) + + on_it, metrics, writer = _observers(args) + eng = Engine(check, strategies.simulated_anneal(step=0.3, seed=args.seed), + max_iterations=args.max_iter, + stop=TargetScore(args.target if args.target is not None else 0.95), + on_iteration=on_it) + res = eng.run(args.start) + if writer: + writer.close() + _report(res.best.state if res.best else res.state, res, metrics, args) + return 0 + + +def cmd_version(_args) -> int: + print(f"cyclework {__version__}") + return 0 + + +def _report(answer, res, metrics: MetricsCollector, args) -> None: + print(f"\nanswer = {answer}") + print(f"status = {res.status.value}") + print(f"metrics = {json.dumps(res.metrics(), default=str)}") + if not args.quiet: + print(f"live = {json.dumps(metrics.snapshot(), default=str)}") + if args.jsonl: + print(f"trace -> {args.jsonl}") + + +def build_parser() -> argparse.ArgumentParser: + # global options live on a parent parser so they work either before OR + # after the subcommand (`cyclework --quiet sqrt 2` and `cyclework sqrt 2 --quiet`). + g = argparse.ArgumentParser(add_help=False) + g.add_argument("--max-iter", type=int, default=50, dest="max_iter") + g.add_argument("--target", type=float, default=None, + help="stop once score >= TARGET (SOLVED)") + g.add_argument("--max-seconds", type=float, default=None, dest="max_seconds", + help="wall-clock stop condition") + g.add_argument("--jsonl", type=str, default=None, help="write JSONL trace to PATH") + g.add_argument("--print-every", type=int, default=1, dest="print_every") + g.add_argument("--quiet", action="store_true") + + p = argparse.ArgumentParser(prog="cyclework", parents=[g], + description="iterative-refinement engine CLI") + sub = p.add_subparsers(dest="command", required=True) + + s = sub.add_parser("sqrt", parents=[g], help="square root via Newton refinement") + s.add_argument("n", type=float) + s.add_argument("--tol", type=float, default=1e-12) + s.set_defaults(func=cmd_sqrt) + + o = sub.add_parser("optimize", parents=[g], help="maximize a bumpy 1-D function") + o.add_argument("--strategy", choices=["hill", "anneal", "grid"], default="anneal") + o.add_argument("--start", type=float, default=0.0) + o.add_argument("--step", type=float, default=0.3) + o.add_argument("--seed", type=int, default=0) + o.add_argument("--restarts", type=int, default=1) + o.add_argument("--range", type=float, nargs=2, default=[-5.0, 5.0]) + o.set_defaults(func=cmd_optimize) + + d = sub.add_parser("demo", parents=[g], help="a guided run with stop conditions + metrics") + d.add_argument("--start", type=float, default=0.0) + d.add_argument("--seed", type=int, default=0) + d.set_defaults(func=cmd_demo) + + v = sub.add_parser("version", parents=[g], help="print version") + v.set_defaults(func=cmd_version) + return p + + +def main(argv: Optional[List[str]] = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + return args.func(args) + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) diff --git a/cyclework/engine.py b/cyclework/engine.py index 80c1f02..a7ef5af 100644 --- a/cyclework/engine.py +++ b/cyclework/engine.py @@ -16,12 +16,27 @@ Convention: `score` is higher-is-better. For a "lower is better" objective (an error/cost), return its negation as the score. + +Two optional, fully backward-compatible extensions can be passed to the engine: + +* ``stop=`` — a :class:`~cyclework.stopping.StopCondition` (or bare callable + ``stop(iteration, history) -> Optional[Status]``) consulted after each + recorded iteration, *after* the built-in ``ok`` check. It halts the run with + whatever terminal status it returns (wall-clock timeout, target score, custom + divergence guard, …). Composable via ``|``. +* ``retry=`` — a :class:`~cyclework.policy.RetryPolicy` that re-attempts a + raising ``check``/``revise`` with backoff before the run aborts as ``ERROR``. + +Both default to the historical behavior (no extra stop, no retries), so existing +callers are unaffected. """ from __future__ import annotations from typing import Any, Callable, Iterator, Optional +from .policy import RetryPolicy +from .stopping import StopCondition, as_stop_condition from .trace import Iteration, Result, Status from .verdict import Verdict @@ -42,11 +57,15 @@ def __init__( min_delta: float = 0.0, seed: Optional[SeedFn] = None, on_iteration: Optional[ObserverFn] = None, + stop: Optional[Any] = None, + retry: Optional[RetryPolicy] = None, ): if max_iterations < 0: raise ValueError("max_iterations must be >= 0") if patience is not None and patience < 1: raise ValueError("patience must be >= 1") + if retry is not None and not isinstance(retry, RetryPolicy): + raise TypeError("retry must be a RetryPolicy") self.check = check self.revise = revise self.max_iterations = max_iterations @@ -54,19 +73,33 @@ def __init__( self.min_delta = min_delta self.seed = seed self.on_iteration = on_iteration + self.stop: Optional[StopCondition] = as_stop_condition(stop) if stop is not None else None + self.retry = retry + + def _call(self, fn, *args): + """Invoke a stage, applying the retry policy if one is configured.""" + if self.retry is None: + return fn(*args) + return self.retry.run(lambda: fn(*args)) # ---- streaming form: yields each iteration as it happens ------------- def cycle(self, initial: Any) -> Iterator[Iteration]: + if self.stop is not None: + self.stop.reset() + history: list[Iteration] = [] state = self.seed(initial) if self.seed else initial for i in range(self.max_iterations): - verdict = self.check(state) + verdict = self._call(self.check, state) it = Iteration(index=i, state=state, verdict=verdict) + history.append(it) if self.on_iteration: self.on_iteration(it) yield it if verdict.ok: return - state = self.revise(state, verdict) + if self.stop is not None and self.stop(it, history) is not None: + return + state = self._call(self.revise, state, verdict) # ---- batch form: runs to completion, returns a Result ---------------- def run(self, initial: Any) -> Result: @@ -75,11 +108,13 @@ def run(self, initial: Any) -> Result: status = Status.EXHAUSTED last_improved: Optional[float] = None stale = 0 + if self.stop is not None: + self.stop.reset() state = self.seed(initial) if self.seed else initial for i in range(self.max_iterations): try: - verdict = self.check(state) + verdict = self._call(self.check, state) except Exception as e: # noqa: BLE001 - surface stage errors as ERROR status return Result(Status.ERROR, state, len(history), history, best, error=f"check failed at iteration {i}: {e}") @@ -94,6 +129,14 @@ def run(self, initial: Any) -> Result: status = Status.SOLVED break + # pluggable stop conditions (wall-clock, target, custom guards). + # consulted after the built-in ok check so SOLVED always wins. + if self.stop is not None: + halt = self.stop(it, history) + if halt is not None: + status = halt + break + # plateau detection on score (higher is better) if self.patience is not None and verdict.score is not None: if last_improved is None or verdict.score > last_improved + self.min_delta: @@ -106,7 +149,7 @@ def run(self, initial: Any) -> Result: break try: - state = self.revise(state, verdict) + state = self._call(self.revise, state, verdict) except Exception as e: # noqa: BLE001 return Result(Status.ERROR, state, len(history), history, best, error=f"revise failed at iteration {i}: {e}") diff --git a/cyclework/observers.py b/cyclework/observers.py new file mode 100644 index 0000000..0ca28a5 --- /dev/null +++ b/cyclework/observers.py @@ -0,0 +1,167 @@ +"""Observers: watch a run as it happens. + +``Engine`` already accepts a single ``on_iteration`` callback. This module turns +that hook into a small, composable ecosystem: + +* :func:`multi` — fan one run out to several observers. +* :class:`JsonlWriter` — append one JSON object per iteration to a stream or + file, giving you a durable, greppable, replayable trace (an "audit log"). +* :class:`MetricsCollector` — accumulate live running stats (best score, count, + improvement events) queryable at any point, without waiting for the Result. +* :class:`ProgressPrinter` — a human-readable one-line-per-iteration printer. + +Every observer is just a ``callable(Iteration) -> None``, so they slot straight +into ``Engine(on_iteration=...)`` and interoperate with any user callback. +""" + +from __future__ import annotations + +import json +import sys +from typing import Any, Callable, IO, List, Optional + +from .trace import Iteration + +Observer = Callable[[Iteration], None] + + +def multi(*observers: Optional[Observer]) -> Observer: + """Combine several observers into one. ``None`` entries are skipped so you + can splice in optional hooks without branching.""" + active: List[Observer] = [o for o in observers if o is not None] + + def _fan(it: Iteration) -> None: + for o in active: + o(it) + + return _fan + + +def _iteration_record(it: Iteration, include_state: bool) -> dict: + rec: dict[str, Any] = { + "index": it.index, + "ok": it.verdict.ok, + "score": it.score, + "feedback": it.verdict.feedback, + } + if it.verdict.detail: + rec["detail"] = _jsonable(it.verdict.detail) + if include_state: + rec["state"] = _jsonable(it.state) + return rec + + +def _jsonable(obj: Any) -> Any: + """Best-effort coercion: keep JSON-native types, stringify the rest.""" + if obj is None or isinstance(obj, (bool, int, float, str)): + return obj + if isinstance(obj, dict): + return {str(k): _jsonable(v) for k, v in obj.items()} + if isinstance(obj, (list, tuple)): + return [_jsonable(v) for v in obj] + return str(obj) + + +class JsonlWriter: + """Write one JSON line per iteration (JSON Lines / newline-delimited JSON). + + Point it at any writable text stream (``sys.stdout``, an open file) or use + :meth:`to_path` to manage a file for you. Flushes each line so the trace is + durable even if the process is killed mid-run. + """ + + def __init__(self, stream: IO[str], include_state: bool = False): + self.stream = stream + self.include_state = include_state + self.count = 0 + + @classmethod + def to_path(cls, path: str, include_state: bool = False) -> "JsonlWriter": + return cls(open(path, "w", encoding="utf-8"), include_state=include_state) + + def __call__(self, it: Iteration) -> None: + rec = _iteration_record(it, self.include_state) + self.stream.write(json.dumps(rec, ensure_ascii=False) + "\n") + self.stream.flush() + self.count += 1 + + def close(self) -> None: + if self.stream not in (sys.stdout, sys.stderr): + self.stream.close() + + def __enter__(self) -> "JsonlWriter": + return self + + def __exit__(self, *exc) -> None: + self.close() + + +class MetricsCollector: + """Accumulate live statistics as iterations stream by. + + Unlike ``Result.metrics`` (available only after the run), this updates in + real time, so a long-running loop can be monitored or a dashboard fed. It + tracks count, best/last/min/max score, and how many iterations strictly + improved on the previous best. + """ + + def __init__(self) -> None: + self.count = 0 + self.scored = 0 + self.best: Optional[float] = None + self.last: Optional[float] = None + self.min: Optional[float] = None + self.max: Optional[float] = None + self.improvements = 0 + self._sum = 0.0 + + def __call__(self, it: Iteration) -> None: + self.count += 1 + s = it.score + if s is None: + return + self.scored += 1 + self.last = s + self._sum += s + self.min = s if self.min is None else min(self.min, s) + self.max = s if self.max is None else max(self.max, s) + if self.best is None or s > self.best: + self.best = s + self.improvements += 1 + + @property + def mean(self) -> Optional[float]: + return (self._sum / self.scored) if self.scored else None + + def snapshot(self) -> dict: + return { + "count": self.count, + "scored": self.scored, + "best": self.best, + "last": self.last, + "min": self.min, + "max": self.max, + "mean": self.mean, + "improvements": self.improvements, + } + + +class ProgressPrinter: + """A tidy human-readable observer: one line per iteration. + + ``every`` throttles output (print every Nth iteration, always printing an + ``ok`` verdict). Writes to ``stream`` (default stdout).""" + + def __init__(self, every: int = 1, stream: IO[str] = sys.stdout, prefix: str = ""): + if every < 1: + raise ValueError("every must be >= 1") + self.every = every + self.stream = stream + self.prefix = prefix + + def __call__(self, it: Iteration) -> None: + if it.verdict.ok or it.index % self.every == 0: + mark = "OK " if it.verdict.ok else "..." + score = "None" if it.score is None else f"{it.score:.6g}" + fb = f" {it.verdict.feedback}" if it.verdict.feedback else "" + self.stream.write(f"{self.prefix}[{it.index:>3}] {mark} score={score}{fb}\n") diff --git a/cyclework/policy.py b/cyclework/policy.py new file mode 100644 index 0000000..c2510ae --- /dev/null +++ b/cyclework/policy.py @@ -0,0 +1,90 @@ +"""Retry / backoff policy for flaky stages. + +Real-world ``check`` and ``revise`` functions call things that fail +transiently — a model endpoint, a compiler, a network probe. Without a policy +a single hiccup aborts the whole run with ``Status.ERROR``. A :class:`RetryPolicy` +lets the engine re-attempt a raising stage a bounded number of times with +exponential backoff (and optional jitter) before giving up. + +The policy is pure and side-effect free apart from sleeping: you hand it the +attempt number and it tells you whether to retry and how long to wait. The +engine owns the loop. ``sleep`` is injectable so tests run instantly. +""" + +from __future__ import annotations + +import random +import time +from typing import Callable, Iterable, Optional, Tuple, Type + + +class RetryPolicy: + def __init__( + self, + max_attempts: int = 3, + base_delay: float = 0.05, + factor: float = 2.0, + max_delay: float = 5.0, + jitter: float = 0.0, + retry_on: Tuple[Type[BaseException], ...] = (Exception,), + sleep: Callable[[float], None] = time.sleep, + rng: Optional[random.Random] = None, + ): + if max_attempts < 1: + raise ValueError("max_attempts must be >= 1") + if base_delay < 0 or max_delay < 0: + raise ValueError("delays must be >= 0") + if factor < 1: + raise ValueError("factor must be >= 1") + if jitter < 0: + raise ValueError("jitter must be >= 0") + if isinstance(retry_on, type): + retry_on = (retry_on,) + self.max_attempts = max_attempts + self.base_delay = float(base_delay) + self.factor = float(factor) + self.max_delay = float(max_delay) + self.jitter = float(jitter) + self.retry_on = tuple(retry_on) + self._sleep = sleep + self._rng = rng or random.Random() + + def delay_for(self, attempt: int) -> float: + """Backoff delay (seconds) before ``attempt`` (1-based). Attempt 1 has + no delay; later attempts grow geometrically, capped at ``max_delay``, + plus up to ``jitter`` seconds of uniform noise.""" + if attempt <= 1: + return 0.0 + raw = self.base_delay * (self.factor ** (attempt - 2)) + raw = min(raw, self.max_delay) + if self.jitter: + raw += self._rng.uniform(0, self.jitter) + return raw + + def should_retry(self, exc: BaseException, attempt: int) -> bool: + """Whether a stage that raised ``exc`` on ``attempt`` (1-based) is worth + retrying: the error type is in ``retry_on`` and attempts remain.""" + return attempt < self.max_attempts and isinstance(exc, self.retry_on) + + def run(self, fn: Callable[[], object]) -> object: + """Call ``fn()`` with retries. Returns its value, or re-raises the last + exception once attempts are exhausted / the error is not retryable.""" + attempt = 0 + while True: + attempt += 1 + try: + return fn() + except BaseException as exc: # noqa: BLE001 - policy decides + if not self.should_retry(exc, attempt): + raise + delay = self.delay_for(attempt + 1) + if delay: + self._sleep(delay) + + def __repr__(self) -> str: # pragma: no cover - cosmetic + return (f"RetryPolicy(max_attempts={self.max_attempts}, " + f"base_delay={self.base_delay}, factor={self.factor})") + + +NO_RETRY = RetryPolicy(max_attempts=1) +"""A policy that never retries — the engine's historical behavior.""" diff --git a/cyclework/search.py b/cyclework/search.py new file mode 100644 index 0000000..87f7082 --- /dev/null +++ b/cyclework/search.py @@ -0,0 +1,120 @@ +"""Keep-best-of-N and random-restart orchestration. + +A single refinement run can land in a poor local optimum. The classic fix is to +run the loop several times from different seeds and keep the best outcome. This +module wraps ``Engine.run`` in that meta-loop: + +* :func:`best_of` — run ``n`` times (optionally over caller-supplied seeds), + return the :class:`BestOf` bundle: the winning :class:`~cyclework.trace.Result`, + every run, and which one won. +* :func:`parallel_best_of` — the same, but evaluate the runs concurrently on a + thread pool. Because each run owns its own state and the engine is pure w.r.t. + shared data, this is safe for I/O-bound ``check``/``revise`` stages. + +Both stop early if a run comes back ``solved`` (configurable), so a cheap lucky +restart short-circuits the expensive ones. +""" + +from __future__ import annotations + +from concurrent.futures import ThreadPoolExecutor +from dataclasses import dataclass, field +from typing import Any, Callable, Iterable, List, Optional + +from .engine import Engine +from .trace import Result, Status + + +@dataclass +class BestOf: + best: Result + runs: List[Result] = field(default_factory=list) + best_index: int = 0 + + @property + def solved(self) -> bool: + return self.best.solved + + def summary(self) -> dict: + scored = [r.best.score for r in self.runs if r.best and r.best.score is not None] + return { + "runs": len(self.runs), + "best_index": self.best_index, + "best_status": self.best.status.value, + "best_score": self.best.best.score if self.best.best else None, + "solved_runs": sum(1 for r in self.runs if r.solved), + "score_spread": (max(scored) - min(scored)) if scored else None, + } + + +def _score_key(r: Result) -> tuple: + """Rank runs: solved first, then by best score (None sorts lowest).""" + solved = 1 if r.solved else 0 + score = r.best.score if (r.best and r.best.score is not None) else float("-inf") + return (solved, score) + + +def _pick_best(results: List[Result]) -> int: + best_i = 0 + for i in range(1, len(results)): + if _score_key(results[i]) > _score_key(results[best_i]): + best_i = i + return best_i + + +def best_of( + engine: Any, + seeds: Iterable[Any], + *, + stop_when_solved: bool = True, +) -> BestOf: + """Run once per value in ``seeds`` (each passed to ``run``), returning the + best outcome. Serial; stops early on the first solved run when + ``stop_when_solved`` is set. + + ``engine`` may be an :class:`Engine` *or* a zero-arg factory returning a + fresh Engine. Pass a factory whenever the engine's ``revise``/stop/observer + hooks are **stateful** (e.g. a strategy from :mod:`cyclework.strategies`), so + each restart begins clean instead of inheriting the previous run's state. + """ + seeds = list(seeds) + if not seeds: + raise ValueError("seeds must be non-empty") + make = engine if callable(engine) and not isinstance(engine, Engine) else (lambda: engine) + runs: List[Result] = [] + for s in seeds: + r = make().run(s) + runs.append(r) + if stop_when_solved and r.solved: + break + best_i = _pick_best(runs) + return BestOf(best=runs[best_i], runs=runs, best_index=best_i) + + +def parallel_best_of( + engine_factory: Callable[[], Engine], + seeds: Iterable[Any], + *, + max_workers: Optional[int] = None, +) -> BestOf: + """Evaluate ``len(seeds)`` runs concurrently on a thread pool and return the + best. + + ``engine_factory`` is called once per run to build a fresh :class:`Engine`, + so stateful stages/strategies/observers are not shared across threads (which + would corrupt their internal state). Ideal when ``check``/``revise`` are + I/O-bound (network, subprocess). All runs complete; there is no early stop + because they run at once. + """ + seeds = list(seeds) + if not seeds: + raise ValueError("seeds must be non-empty") + + def _run(seed: Any) -> Result: + return engine_factory().run(seed) + + with ThreadPoolExecutor(max_workers=max_workers) as pool: + runs = list(pool.map(_run, seeds)) + + best_i = _pick_best(runs) + return BestOf(best=runs[best_i], runs=runs, best_index=best_i) diff --git a/cyclework/stopping.py b/cyclework/stopping.py new file mode 100644 index 0000000..169fb35 --- /dev/null +++ b/cyclework/stopping.py @@ -0,0 +1,214 @@ +"""Pluggable stop conditions. + +The engine has always known three ways to stop: a verdict came back ``ok``, +the score plateaued, or the iteration budget ran out. Those are baked into +``Engine.run``. This module makes stopping *composable* and *domain-driven* +without touching that core: a ``StopCondition`` is any callable + + stop(iteration, history) -> Optional[Status] + +that returns a terminal :class:`~cyclework.trace.Status` to halt the run (with +that status) or ``None`` to keep going. It is consulted *after* an iteration is +recorded and *after* the built-in ``ok`` check, so ``SOLVED`` always wins. + +Conditions compose with ``|`` (first one that fires wins) and the ready-made +ones cover the common asks: wall-clock deadline, target score reached, and +absolute/relative no-improvement patience. They are deliberately independent of +the engine so you can also use them to drive a hand-rolled ``cycle()`` loop. +""" + +from __future__ import annotations + +import time +from typing import Callable, List, Optional + +from .trace import Iteration, Status + +StopFn = Callable[[Iteration, List[Iteration]], Optional[Status]] + + +class StopCondition: + """Wraps a ``stop(iteration, history) -> Optional[Status]`` callable. + + Instances are callable and combine with ``|``: + + stop = MaxSeconds(30) | TargetScore(0.99) + """ + + def __init__(self, fn: StopFn, name: str = "custom"): + if not callable(fn): + raise TypeError("StopCondition needs a callable stop(it, history)") + self._fn = fn + self.name = name + + def __call__(self, it: Iteration, history: List[Iteration]) -> Optional[Status]: + return self._fn(it, history) + + def __or__(self, other: "StopCondition") -> "StopCondition": + other = as_stop_condition(other) + return _Any([self, other]) + + def reset(self) -> None: + """Clear any per-run state. Called by the engine at the start of a run + so a single StopCondition object is safe to reuse across runs.""" + + def __repr__(self) -> str: # pragma: no cover - cosmetic + return f"StopCondition({self.name})" + + +class _Any(StopCondition): + """Fires as soon as any child condition fires (short-circuit, in order).""" + + def __init__(self, children: List[StopCondition]): + # flatten nested _Any for a clean, order-preserving chain + flat: List[StopCondition] = [] + for c in children: + if isinstance(c, _Any): + flat.extend(c.children) + else: + flat.append(c) + self.children = flat + super().__init__(self._eval, name=" | ".join(c.name for c in flat)) + + def _eval(self, it: Iteration, history: List[Iteration]) -> Optional[Status]: + for c in self.children: + status = c(it, history) + if status is not None: + return status + return None + + def reset(self) -> None: + for c in self.children: + c.reset() + + +def as_stop_condition(obj) -> StopCondition: + """Coerce a bare callable into a StopCondition (idempotent).""" + if isinstance(obj, StopCondition): + return obj + if callable(obj): + return StopCondition(obj, name=getattr(obj, "__name__", "custom")) + raise TypeError("stop must be a StopCondition or a callable(it, history)") + + +# --------------------------------------------------------------------------- +# Ready-made conditions +# --------------------------------------------------------------------------- + +class MaxSeconds(StopCondition): + """Stop once ``seconds`` of wall-clock time have elapsed since the run began. + + Returns :attr:`Status.TIMEOUT`. The clock starts at :meth:`reset` (which the + engine calls before iterating) so the object is reusable. + """ + + def __init__(self, seconds: float, clock: Callable[[], float] = time.monotonic): + if seconds < 0: + raise ValueError("seconds must be >= 0") + self.seconds = float(seconds) + self._clock = clock + self._start = clock() + super().__init__(self._eval, name=f"max_seconds={seconds}") + + def reset(self) -> None: + self._start = self._clock() + + def _eval(self, it: Iteration, history: List[Iteration]) -> Optional[Status]: + if self._clock() - self._start >= self.seconds: + return Status.TIMEOUT + return None + + +class TargetScore(StopCondition): + """Stop once an iteration's score reaches ``target`` (higher is better). + + Returns :attr:`Status.SOLVED` — reaching the target is success even if the + verdict never set ``ok=True``. Unscored iterations are ignored. + """ + + def __init__(self, target: float): + self.target = float(target) + super().__init__(self._eval, name=f"target_score>={target}") + + def _eval(self, it: Iteration, history: List[Iteration]) -> Optional[Status]: + if it.score is not None and it.score >= self.target: + return Status.SOLVED + return None + + +class NoImprovement(StopCondition): + """Stop after ``patience`` consecutive iterations without improving the best + score seen. This mirrors the engine's built-in plateau logic but as a + reusable, composable object and with an optional *relative* tolerance. + + improvement is counted when:: + + score > best_so_far + max(min_delta, best_so_far * rel_delta) + + Returns :attr:`Status.PLATEAU`. + """ + + def __init__(self, patience: int, min_delta: float = 0.0, rel_delta: float = 0.0): + if patience < 1: + raise ValueError("patience must be >= 1") + if min_delta < 0 or rel_delta < 0: + raise ValueError("deltas must be >= 0") + self.patience = patience + self.min_delta = float(min_delta) + self.rel_delta = float(rel_delta) + self._best: Optional[float] = None + self._stale = 0 + super().__init__(self._eval, name=f"no_improvement(patience={patience})") + + def reset(self) -> None: + self._best = None + self._stale = 0 + + def _eval(self, it: Iteration, history: List[Iteration]) -> Optional[Status]: + s = it.score + if s is None: + return None + if self._best is None: + self._best = s + self._stale = 0 + return None + threshold = self._best + max(self.min_delta, abs(self._best) * self.rel_delta) + if s > threshold: + self._best = s + self._stale = 0 + else: + self._stale += 1 + if self._stale >= self.patience: + return Status.PLATEAU + return None + + +class ScoreThreshold(StopCondition): + """Stop (with a caller-chosen status, default ABORTED) when the score falls + *below* ``floor`` — a divergence / bail-out guard for objectives that can + blow up. Unscored iterations are ignored.""" + + def __init__(self, floor: float, status: Status = Status.ABORTED): + self.floor = float(floor) + self.status = status + super().__init__(self._eval, name=f"score_floor={floor}") + + def _eval(self, it: Iteration, history: List[Iteration]) -> Optional[Status]: + if it.score is not None and it.score < self.floor: + return self.status + return None + + +class Predicate(StopCondition): + """Stop when ``pred(state)`` is truthy. A convenience for domain rules that + look only at the candidate. Returns ``status`` (default SOLVED).""" + + def __init__(self, pred: Callable, status: Status = Status.SOLVED, name: str = "predicate"): + if not callable(pred): + raise TypeError("Predicate needs a callable(state)") + self.pred = pred + self.status = status + super().__init__(self._eval, name=name) + + def _eval(self, it: Iteration, history: List[Iteration]) -> Optional[Status]: + return self.status if self.pred(it.state) else None diff --git a/cyclework/strategies.py b/cyclework/strategies.py new file mode 100644 index 0000000..2c3c190 --- /dev/null +++ b/cyclework/strategies.py @@ -0,0 +1,139 @@ +"""A small library of ready-made ``revise`` strategies. + +The engine is agnostic about *how* you produce the next candidate — that is the +``revise(state, verdict)`` function. Writing a good one is often the hard part, +so this module supplies reusable, domain-independent search strategies as +``revise`` factories. Each returns a function you can hand straight to +``Engine(check, revise=...)``. + +Because these are ordinary revise functions they compose with every other +feature — stop conditions, retries, observers, best-of restarts. + +Two families are provided: + +* **numeric local search** over a scalar ``x``: :func:`hill_climb`, + :func:`simulated_anneal`, :func:`grid_step`. These read the verdict's + ``score`` to decide direction/acceptance. +* **generic neighbor search**: :func:`generic_hill_climb`, which works on *any* + state given a ``neighbors(state)`` generator and the verdict score. + +All randomized strategies accept a ``seed`` / ``rng`` for reproducibility. +""" + +from __future__ import annotations + +import math +import random +from typing import Any, Callable, Iterable, List, Optional + +from .verdict import Verdict + +ReviseFn = Callable[[Any, Verdict], Any] + + +def hill_climb(step: float = 1.0, shrink: float = 0.5, grow: float = 1.0) -> ReviseFn: + """Scalar hill-climb using the sign of score change. + + Tries ``x + step``; the *next* verdict tells us (via score) whether that + helped. Since a revise function only sees the current state and verdict, we + remember the last move on the returned closure and reverse+shrink it when + the score drops, or keep+grow it when the score rises. Requires scored + verdicts. + """ + state = {"dir": 1.0, "step": step, "last_score": None} + + def revise(x: float, v: Verdict) -> float: + s = v.score + if s is None: + raise ValueError("hill_climb requires a scored verdict") + last = state["last_score"] + if last is not None: + if s > last: + state["step"] *= grow + else: + state["dir"] *= -1.0 + state["step"] *= shrink + state["last_score"] = s + return x + state["dir"] * state["step"] + + return revise + + +def simulated_anneal( + step: float = 1.0, + t0: float = 1.0, + cooling: float = 0.9, + seed: Optional[int] = None, +) -> ReviseFn: + """Simulated-annealing revise for a scalar ``x`` (score higher is better). + + Proposes ``x ± step`` and accepts an uphill move always, a downhill move + with probability ``exp(delta / T)``; temperature ``T`` cools by ``cooling`` + each step. Escapes local optima the plain hill-climber gets stuck in. + """ + rng = random.Random(seed) + state = {"T": t0, "last_score": None, "last_x": None} + + def revise(x: float, v: Verdict) -> float: + s = v.score + if s is None: + raise ValueError("simulated_anneal requires a scored verdict") + last_s, last_x = state["last_score"], state["last_x"] + cur = x + if last_s is not None and last_x is not None: + delta = s - last_s + if delta < 0 and state["T"] > 1e-12: + if rng.random() >= math.exp(delta / state["T"]): + cur = last_x # reject: revert to the previous point + state["last_score"] = s + state["last_x"] = cur + state["T"] *= cooling + return cur + rng.choice((-1.0, 1.0)) * step + + return revise + + +def grid_step(values: Iterable[float]) -> ReviseFn: + """Sweep a fixed grid of candidate values in order, one per iteration. + + Ignores the score for *movement* (the engine's best-tracking still keeps the + top scorer). Once the grid is exhausted it holds the last value, so pair it + with ``max_iterations=len(values)`` or a stop condition. + """ + grid: List[float] = list(values) + if not grid: + raise ValueError("grid must be non-empty") + idx = {"i": 0} + + def revise(x: float, v: Verdict) -> float: + idx["i"] += 1 + return grid[min(idx["i"], len(grid) - 1)] + + return revise + + +def generic_hill_climb( + neighbors: Callable[[Any], Iterable[Any]], + score_of: Callable[[Any], float], + seed: Optional[int] = None, +) -> ReviseFn: + """Neighbor-search hill-climb over *any* state type. + + ``neighbors(state)`` yields candidate successors; ``score_of(state)`` scores + one (higher better). revise returns the best-scoring neighbor, or a random + one if none beats the incumbent (a light random-walk kick). Works on tuples, + strings, graphs — anything with a neighbor function. + """ + rng = random.Random(seed) + + def revise(state: Any, v: Verdict) -> Any: + cands = list(neighbors(state)) + if not cands: + return state + incumbent = score_of(state) + best = max(cands, key=score_of) + if score_of(best) > incumbent: + return best + return rng.choice(cands) + + return revise diff --git a/cyclework/trace.py b/cyclework/trace.py index 138a91c..b6ae1e9 100644 --- a/cyclework/trace.py +++ b/cyclework/trace.py @@ -17,10 +17,18 @@ class Status(str, Enum): - SOLVED = "solved" # a verdict came back ok=True + SOLVED = "solved" # a verdict came back ok=True (or a target was hit) PLATEAU = "plateau" # score stopped improving (patience exhausted) EXHAUSTED = "exhausted" # hit max_iterations without solving ERROR = "error" # a stage raised and the run was aborted + TIMEOUT = "timeout" # a wall-clock stop condition fired + ABORTED = "aborted" # a custom stop condition halted the run + + @property + def terminal(self) -> bool: + """Every Status is terminal (a run ends in exactly one). Provided so + callers can assert intent without special-casing the enum.""" + return True @dataclass @@ -56,3 +64,48 @@ def summary(self) -> dict: "best_score": self.best.score if self.best else None, "error": self.error, } + + def metrics(self) -> dict: + """A richer, quantitative view of the run than :meth:`summary`. + + Adds score aggregates (min/max/mean/last), total absolute improvement + from first to best score, and the index at which the best was found — + the numbers you want when comparing strategies or tuning budgets. All + score-derived fields are ``None`` when no iteration was scored. + """ + scores = [it.score for it in self.history if it.score is not None] + best_index = self.best.index if self.best is not None else None + first = scores[0] if scores else None + improvement = (self.best.score - first) if (scores and self.best and self.best.score is not None) else None + base = self.summary() + base.update({ + "scored_iterations": len(scores), + "min_score": min(scores) if scores else None, + "max_score": max(scores) if scores else None, + "mean_score": (sum(scores) / len(scores)) if scores else None, + "best_index": best_index, + "total_improvement": improvement, + }) + return base + + def to_dict(self, include_states: bool = False) -> dict: + """A JSON-serializable snapshot of the whole run, trace included. + + States are stringified (or omitted) since a candidate can be anything; + set ``include_states=True`` to keep ``str(state)`` per iteration. + """ + def row(it: "Iteration") -> dict: + d = { + "index": it.index, + "ok": it.verdict.ok, + "score": it.score, + "feedback": it.verdict.feedback, + } + if include_states: + d["state"] = str(it.state) + return d + + return { + "metrics": self.metrics(), + "history": [row(it) for it in self.history], + } diff --git a/demos/06_wallclock_deadline.py b/demos/06_wallclock_deadline.py new file mode 100644 index 0000000..a65c20a --- /dev/null +++ b/demos/06_wallclock_deadline.py @@ -0,0 +1,36 @@ +"""Scenario 6 - wall-clock deadlines with a composable stop condition. + +Some loops must respect a real-time budget: "refine for at most 200ms, then hand +back the best you have." `MaxSeconds` is a pluggable StopCondition that halts the +run with Status.TIMEOUT once the deadline passes - and it composes with a +TargetScore via `|`, so the run stops on whichever comes first. +""" +from _common import Engine, Verdict, rule, show_summary + +from cyclework import MaxSeconds, TargetScore + + +def main() -> None: + rule("WALL-CLOCK DEADLINE - stop on time OR on target, whichever first") + + # a slow-converging loop: each step halves the remaining error + def check(x): + err = abs(x - 100.0) + return Verdict(ok=err < 1e-9, score=-err, detail={"err": err}) + + def revise(x, _v): + return x + (100.0 - x) * 0.05 # deliberately slow + + stop = MaxSeconds(0.05) | TargetScore(-0.5) + print(f"\nstop = {stop.name}") + res = Engine(check, revise, max_iterations=100000, stop=stop).run(0.0) + + show_summary(res) + print(f"\n -> the run ended with status={res.status.value!r}; " + f"best candidate x={res.best.state:.4f}") + print(" Either the target score or the time budget ended it - the caller " + "never had to hand-roll a timer.") + + +if __name__ == "__main__": + main() diff --git a/demos/07_retry_backoff.py b/demos/07_retry_backoff.py new file mode 100644 index 0000000..a99564c --- /dev/null +++ b/demos/07_retry_backoff.py @@ -0,0 +1,49 @@ +"""Scenario 7 - retry/backoff for flaky stages. + +A `check` or `revise` that calls a network service, a model, or a subprocess +fails transiently. Without a policy a single blip aborts the whole run as +Status.ERROR. A RetryPolicy re-attempts a raising stage with exponential backoff +before giving up - turning a fragile loop into a resilient one, with no change to +the stage's own logic. +""" +from _common import Engine, Verdict, rule, show_summary + +from cyclework import RetryPolicy + + +def main() -> None: + rule("RETRY / BACKOFF - survive transient stage failures") + + # a check that is 'flaky' the first two times it sees each new state + seen = {} + + def check(x): + seen[x] = seen.get(x, 0) + 1 + if seen[x] < 2: + raise ConnectionError(f"transient blip evaluating x={x}") + err = abs(x - 42) + return Verdict(ok=err == 0, score=-err) + + def revise(x, _v): + return x + 1 + + # sleep is injected as a no-op so the demo is instant; real use would sleep. + policy = RetryPolicy(max_attempts=4, base_delay=0.01, sleep=lambda s: None) + print(f"\npolicy = {policy}") + print("delays for attempts 1..4:", + [round(policy.delay_for(a), 3) for a in range(1, 5)]) + + res = Engine(check, revise, max_iterations=60, retry=policy).run(30) + show_summary(res) + + print("\nWithout the policy the first ConnectionError would have aborted the " + "run as ERROR; with it, each blip is absorbed and the loop converges.") + + # contrast: same flaky check, no policy -> ERROR + seen.clear() + bad = Engine(check, revise, max_iterations=60).run(30) + print(f" (no-policy control run -> status={bad.status.value}: {bad.error})") + + +if __name__ == "__main__": + main() diff --git a/demos/08_jsonl_audit_trace.py b/demos/08_jsonl_audit_trace.py new file mode 100644 index 0000000..daae36c --- /dev/null +++ b/demos/08_jsonl_audit_trace.py @@ -0,0 +1,49 @@ +"""Scenario 8 - a durable JSONL audit trace + live metrics. + +Beyond the in-memory Result.history, cyclework can stream each iteration to disk +as newline-delimited JSON (JSONL): a greppable, replayable audit log that +survives a crash. A MetricsCollector accumulates live stats alongside it, and +`multi()` fans the run out to both observers at once. +""" +import json +import os +import tempfile + +from _common import Engine, Verdict, rule, show_summary + +from cyclework import JsonlWriter, MetricsCollector, multi + + +def main() -> None: + rule("JSONL AUDIT TRACE - durable, replayable, plus live metrics") + + def check(x): + loss = (x - 3) ** 2 + 1 + return Verdict(ok=loss <= 1.0 + 1e-9, score=-loss, feedback=f"loss={loss:.3f}") + + def revise(x, _v): + return x + (3 - x) * 0.4 + + path = os.path.join(tempfile.gettempdir(), "cyclework_demo_trace.jsonl") + writer = JsonlWriter.to_path(path, include_state=True) + metrics = MetricsCollector() + + res = Engine(check, revise, max_iterations=40, + on_iteration=multi(writer, metrics)).run(0.0) + writer.close() + + show_summary(res) + print(f"\n wrote {writer.count} JSONL records -> {path}") + print("\n first 3 lines of the audit trace:") + with open(path, encoding="utf-8") as f: + for line in list(f)[:3]: + rec = json.loads(line) + print(f" idx={rec['index']} score={rec['score']:.4f} " + f"state={rec['state']:.4f} feedback={rec['feedback']!r}") + + print(f"\n live metrics snapshot: {json.dumps(metrics.snapshot(), default=str)}") + os.remove(path) + + +if __name__ == "__main__": + main() diff --git a/demos/09_strategy_library.py b/demos/09_strategy_library.py new file mode 100644 index 0000000..6b86c25 --- /dev/null +++ b/demos/09_strategy_library.py @@ -0,0 +1,47 @@ +"""Scenario 9 - the revise strategy library. + +Writing a good `revise` is often the hard part. cyclework.strategies ships +reusable, domain-independent search strategies you can drop straight in: +hill-climb, simulated annealing, and a grid sweep. Here all three attack the same +bumpy 1-D objective so you can see how they differ. +""" +import math + +from _common import Engine, Verdict, rule + +from cyclework import strategies + + +def bumpy(x: float) -> float: + # multi-modal: a global max near x=2, plus deceptive local bumps + return math.sin(3 * x) - 0.1 * (x - 2) ** 2 + + +def main() -> None: + rule("STRATEGY LIBRARY - hill-climb vs anneal vs grid on a bumpy function") + + def check(x): + return Verdict(ok=False, score=bumpy(x)) + + runs = { + "hill_climb ": Engine(check, strategies.hill_climb(step=0.2), + max_iterations=200).run(-4.0), + "sim_anneal ": Engine(check, strategies.simulated_anneal(step=0.3, seed=1), + max_iterations=200).run(-4.0), + "grid_step ": Engine(check, strategies.grid_step( + [-5 + 10 * i / 199 for i in range(200)]), + max_iterations=200).run(-5.0), + } + + print("\n objective f(x) = sin(3x) - 0.1(x-2)^2 (start x=-4)\n") + for name, res in runs.items(): + b = res.best + print(f" {name}: best x={b.state:8.4f} f(x)={b.score:.4f} " + f"(scored {res.metrics()['scored_iterations']} candidates)") + + print("\n Hill-climb is fast but can stall on a local bump; annealing takes " + "downhill risks to escape it; grid is exhaustive but budget-bound.") + + +if __name__ == "__main__": + main() diff --git a/demos/10_best_of_restarts.py b/demos/10_best_of_restarts.py new file mode 100644 index 0000000..b5bc51f --- /dev/null +++ b/demos/10_best_of_restarts.py @@ -0,0 +1,46 @@ +"""Scenario 10 - keep-best-of-N random restarts. + +One run can land in a poor local optimum. best_of() / parallel_best_of() run the +loop from several seeds and keep the winner - the classic random-restart cure. +Because the hill-climb strategy is *stateful*, we hand the searcher a *factory* +that builds a fresh engine per restart (via parallel_best_of), so no run +contaminates another. +""" +import math + +from _common import Engine, Verdict, rule + +from cyclework import parallel_best_of, strategies + + +def bumpy(x: float) -> float: + return math.sin(3 * x) - 0.1 * (x - 2) ** 2 + + +def main() -> None: + rule("KEEP-BEST-OF-N - random restarts beat a single stuck run") + + def check(x): + return Verdict(ok=False, score=bumpy(x)) + + def factory() -> Engine: + # a *fresh* stateful hill-climber per restart + return Engine(check, strategies.hill_climb(step=0.2), max_iterations=80) + + # single run from a deceptive start + single = factory().run(-4.5) + print(f"\n single run from x=-4.5 -> best f(x)={single.best.score:.4f} " + f"at x={single.best.state:.4f}") + + seeds = [-4.5, -2.0, 0.5, 2.0, 3.5] + bundle = parallel_best_of(factory, seeds, max_workers=4) + b = bundle.best.best + print(f" best-of-{len(seeds)} restarts -> best f(x)={b.score:.4f} " + f"at x={b.state:.4f} (won by seed #{bundle.best_index})") + print(f"\n bundle summary: {bundle.summary()}") + print("\n Restarts trade a little more compute for robustness against bad " + "starting points - no change to the loop itself.") + + +if __name__ == "__main__": + main() diff --git a/demos/11_checkpoint_resume.py b/demos/11_checkpoint_resume.py new file mode 100644 index 0000000..02bae53 --- /dev/null +++ b/demos/11_checkpoint_resume.py @@ -0,0 +1,54 @@ +"""Scenario 11 - checkpoint & resume. + +A long or fragile run shouldn't restart from scratch if the process dies. A +Checkpoint is an observer that atomically persists the current candidate (and the +best seen) to disk every N iterations. A later invocation loads it and resumes +from that state instead of the original seed. + +This demo simulates a crash: run 1 stops after a few iterations; run 2 resumes +from the checkpoint and finishes the job. +""" +import os +import tempfile + +from _common import Engine, Verdict, rule, show_summary + +from cyclework import Checkpoint + + +def main() -> None: + rule("CHECKPOINT / RESUME - survive a crash mid-run") + + def check(x): + # climb integers toward the goal + return Verdict(ok=x >= 10, score=float(x), feedback=f"at {x}") + + def revise(x, _v): + return x + 1 + + path = os.path.join(tempfile.gettempdir(), "cyclework_demo_ckpt.json") + if os.path.exists(path): + os.remove(path) + + # --- run 1: "crashes" after a small budget -------------------------------- + cp1 = Checkpoint(path, every=1) + r1 = Engine(check, revise, max_iterations=4, on_iteration=cp1).run(0) + print("\n run 1 (budget=4, simulated crash):") + show_summary(r1) + saved = cp1.load() + print(f" checkpoint on disk -> index={saved.index} state={saved.state} " + f"best_score={saved.best_score}") + + # --- run 2: resume from the checkpoint instead of the original seed ------- + resume_from = Checkpoint.resume_state(path, default=0) + print(f"\n run 2 resumes from state={resume_from} (not the original seed 0):") + cp2 = Checkpoint(path, every=1) + r2 = Engine(check, revise, max_iterations=20, on_iteration=cp2).run(resume_from) + show_summary(r2) + print(f"\n Finished at state={r2.state} with status={r2.status.value}. " + "The work from run 1 was not thrown away.") + os.remove(path) + + +if __name__ == "__main__": + main() diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index c3a73ea..5f34101 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -5,8 +5,17 @@ candidate, check it against a goal, use the feedback to produce a better one, repeat until it's good enough, stops improving, or you run out of budget. This document explains how the few pieces fit together. -The whole library is ~150 lines of pure standard-library Python across four -modules. There is no runtime dependency, no I/O, and no global state. +The core engine is ~150 lines of pure standard-library Python; the 0.2 +capability modules (stop conditions, retry, observers, restarts, checkpointing, +strategies, CLI) add more, all still pure standard library. There is no runtime +dependency and no global state. The engine itself never touches the network; the +only optional I/O lives in explicitly opt-in observers (`JsonlWriter`, +`Checkpoint`) you have to wire in yourself. + +The design rule for 0.2 was **additive and composable**: every new piece plugs +into the existing `stop=`, `retry=`, or `on_iteration=` seams (or wraps +`Engine.run` from the outside, as the restart/CLI layers do) without changing the +core loop's contract, so all pre-0.2 callers keep working unchanged. ## The loop @@ -62,6 +71,8 @@ The refinement engine. Constructed with `check`, `revise`, and optional knobs: | `patience` / `min_delta` | plateau detection: stop after `patience` checks with no score gain of at least `min_delta` | | `seed` | one-time preprocess of the initial candidate | | `on_iteration` | observer callback, fired with every recorded `Iteration` | +| `stop` | a composable `StopCondition` (or bare callable) consulted after each iteration, *after* the built-in `ok` check | +| `retry` | a `RetryPolicy` that re-attempts a raising `check`/`revise` with backoff before the run aborts as `ERROR` | It exposes two forms of the same loop: @@ -108,13 +119,26 @@ classDiagram ``` ### `Status` (`cyclework/trace.py`) -The four terminal states a run can reach — and the whole point of making the -loop a first-class object is that you can tell them apart: +The terminal states a run can reach — and the whole point of making the loop a +first-class object is that you can tell them apart: -- **`SOLVED`** — a verdict came back `ok=True`. +- **`SOLVED`** — a verdict came back `ok=True` (or a `TargetScore` / `Predicate` stop condition fired). - **`PLATEAU`** — the score stopped improving (patience exhausted). - **`EXHAUSTED`** — `max_iterations` reached without solving. - **`ERROR`** — a stage raised; the run aborted with the trace so far. +- **`TIMEOUT`** — a wall-clock (`MaxSeconds`) stop condition fired. +- **`ABORTED`** — a custom stop condition (e.g. a `ScoreThreshold` divergence guard) halted the run. + +### 0.2 capability modules +Each plugs into an existing engine seam and is independently usable: + +- **`stopping.py`** — composable `StopCondition`s (`MaxSeconds`, `TargetScore`, `NoImprovement`, `ScoreThreshold`, `Predicate`) that combine with `|`; passed as `stop=`. +- **`policy.py`** — `RetryPolicy` (bounded exponential backoff + optional jitter, injectable `sleep`) passed as `retry=`. +- **`observers.py`** — `on_iteration` sinks: `multi` (fan-out), `JsonlWriter` (durable JSONL audit trace), `MetricsCollector` (live stats), `ProgressPrinter`. +- **`search.py`** — `best_of` / `parallel_best_of` keep-best-of-N restarts wrapping `Engine.run`, returning a `BestOf` bundle. +- **`checkpoint.py`** — `Checkpoint`, an observer that atomically persists state for crash-resume. +- **`strategies.py`** — ready-made `revise` factories (`hill_climb`, `simulated_anneal`, `grid_step`, `generic_hill_climb`). +- **`cli.py` / `__main__.py`** — `python -m cyclework` front-end exercising the above on self-contained problems. ### `examples` (`cyclework/examples.py`) Worked uses of the engine, each exercised by the tests: `sqrt_newton` (a numeric diff --git a/docs/DEMOS.md b/docs/DEMOS.md index eefef3e..f54ee14 100644 --- a/docs/DEMOS.md +++ b/docs/DEMOS.md @@ -1,12 +1,14 @@ # Demos -Five runnable scenarios in [`../demos/`](../demos/), each targeting a different +Eleven runnable scenarios in [`../demos/`](../demos/), each targeting a different audience. Every scenario drives the **real** cyclework API (`Engine`, -`Verdict`, the `Result`/`Iteration` trace), touches no network, prints narrated -output, and exits 0 — so they double as smoke tests of the public surface. +`Verdict`, the `Result`/`Iteration` trace, plus the 0.2 stop conditions, retry +policy, observers, restarts, checkpointing, and strategy library), touches no +network, prints narrated output, and exits 0 — so they double as smoke tests of +the public surface. ```bash -python demos/run_all.py # all five, end to end +python demos/run_all.py # all of them, end to end python demos/02_numeric_solvers.py # or just one ``` @@ -20,6 +22,12 @@ cleanly. | 3 | [`03_plateau_and_budget.py`](../demos/03_plateau_and_budget.py) | SREs / production loop owners | Honest stopping: the engine driven into all four terminal states (SOLVED / PLATEAU / EXHAUSTED / ERROR) on purpose, always keeping the best candidate seen. | | 4 | [`04_feedback_refiner.py`](../demos/04_feedback_refiner.py) | Data / content pipeline engineers | Feedback-driven revision: `Verdict.feedback` names the first broken rule and `revise` applies exactly that fix, normalizing messy text into a URL slug one cycle at a time. | | 5 | [`05_streaming_observability.py`](../demos/05_streaming_observability.py) | Observability / platform engineers | Watching a loop live: the `on_iteration` push callback (a live progress bar) and the `Engine.cycle()` pull generator that lets the caller stop early. | +| 6 | [`06_wallclock_deadline.py`](../demos/06_wallclock_deadline.py) | Production loop owners | Composable stop conditions: a `MaxSeconds` wall-clock deadline OR'd (`|`) with a `TargetScore`, whichever trips first, halting with the matching status. | +| 7 | [`07_retry_backoff.py`](../demos/07_retry_backoff.py) | Reliability engineers | A `RetryPolicy` absorbing a flaky stage's transient failures with bounded exponential backoff, so a hiccup no longer aborts the whole run. | +| 8 | [`08_jsonl_audit_trace.py`](../demos/08_jsonl_audit_trace.py) | Observability / compliance | A durable JSONL audit trace (`JsonlWriter`) plus a live `MetricsCollector`, both fed from one run via `multi(...)`. | +| 9 | [`09_strategy_library.py`](../demos/09_strategy_library.py) | Optimization users | Dropping in ready-made `revise` strategies — `hill_climb`, `simulated_anneal`, `grid_step` — with no engine changes. | +| 10 | [`10_best_of_restarts.py`](../demos/10_best_of_restarts.py) | Optimization users | Random restarts with `best_of` escaping a local optimum a single run gets stuck in, and reporting the winning run. | +| 11 | [`11_checkpoint_resume.py`](../demos/11_checkpoint_resume.py) | Long-run / fault-tolerance | Checkpointing candidate state atomically and resuming from it after a simulated crash instead of restarting from the seed. | ## The shape they share @@ -36,4 +44,5 @@ Every demo is the same loop with a different `check`/`revise` pair: Each demo prints clear, narrated output and exits 0, so they double as smoke tests — [`tests/test_demos.py`](../tests/test_demos.py) runs every one under -`pytest`, alongside the unit tests for the engine and examples. +`pytest`, alongside the unit tests for the engine, stop conditions, retry +policy, observers, restarts, checkpointing, strategies, and CLI. diff --git a/pyproject.toml b/pyproject.toml index 97763c1..64c975a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "cyclework" -version = "0.1.0" +version = "0.2.0" description = "A tiny, dependency-free engine for iterative refinement loops: propose, check, revise, repeat." readme = "README.md" requires-python = ">=3.9" @@ -20,6 +20,9 @@ dependencies = [] [project.optional-dependencies] dev = ["pytest>=7"] +[project.scripts] +cyclework = "cyclework.cli:main" + [project.urls] Homepage = "https://github.com/cognis-digital/cyclework" diff --git a/tests/test_checkpoint.py b/tests/test_checkpoint.py new file mode 100644 index 0000000..993d7ea --- /dev/null +++ b/tests/test_checkpoint.py @@ -0,0 +1,94 @@ +"""Tests for checkpoint / resume.""" + +import json +import os + +import pytest + +from cyclework import Engine, Verdict, Checkpoint, CheckpointData + + +def test_checkpoint_writes_and_loads(tmp_path): + p = tmp_path / "ck.json" + cp = Checkpoint(str(p), every=1) + eng = Engine(lambda x: Verdict(ok=x >= 3, score=float(x)), + lambda s, v: s + 1, max_iterations=10, on_iteration=cp) + res = eng.run(0) + assert res.solved + data = cp.load() + assert isinstance(data, CheckpointData) + assert data.index == res.iterations - 1 + assert data.state == 3 + assert data.best_score == 3.0 + assert data.best_state == 3 + + +def test_checkpoint_every_throttles_writes(tmp_path): + p = tmp_path / "ck.json" + cp = Checkpoint(str(p), every=5) + eng = Engine(lambda x: Verdict(ok=False, score=float(x)), + lambda s, v: s + 1, max_iterations=12, on_iteration=cp) + eng.run(0) + # indices 0,5,10 -> 3 writes + assert cp.writes == 3 + + +def test_resume_state_from_checkpoint(tmp_path): + p = tmp_path / "ck.json" + # first run: stop early via max_iterations, leaving progress at state 2 + cp = Checkpoint(str(p), every=1) + Engine(lambda x: Verdict(ok=False, score=float(x)), + lambda s, v: s + 1, max_iterations=3, on_iteration=cp).run(0) + # resume: pick up from the checkpointed state instead of the default seed + resumed = Checkpoint.resume_state(str(p), default=0) + assert resumed == 2 # last recorded candidate index in a 3-iter run + + +def test_resume_state_default_when_missing(tmp_path): + p = tmp_path / "nope.json" + assert Checkpoint.resume_state(str(p), default=42) == 42 + + +def test_checkpoint_load_none_when_absent(tmp_path): + cp = Checkpoint(str(tmp_path / "absent.json")) + assert cp.load() is None + + +def test_checkpoint_atomic_no_tmp_left(tmp_path): + p = tmp_path / "ck.json" + cp = Checkpoint(str(p), every=1) + cp(_it(0, 1.0)) + leftovers = [f for f in os.listdir(tmp_path) if f.endswith(".tmp")] + assert leftovers == [] + + +def test_checkpoint_json_roundtrip_content(tmp_path): + p = tmp_path / "ck.json" + cp = Checkpoint(str(p), every=1) + cp(_it(4, 2.0, state={"key": "value"})) + raw = json.loads(p.read_text(encoding="utf-8")) + assert raw["index"] == 4 + assert raw["state"] == {"key": "value"} + + +def test_checkpoint_pickle_codec_for_arbitrary_state(tmp_path): + import pickle + p = tmp_path / "ck.pkl" + cp = Checkpoint(str(p), every=1, dump=pickle.dumps, load=pickle.loads) + obj = ("tuple", 1, 2.0, {"nested": [1, 2]}) + cp(_it(0, 1.0, state=obj)) + data = cp.load() + assert data.state == obj + + +def test_checkpoint_bad_every(): + with pytest.raises(ValueError): + Checkpoint("x", every=0) + + +# helpers ------------------------------------------------------------------- + +def _it(index, score, state=None): + from cyclework import Iteration, Verdict + return Iteration(index=index, state=index if state is None else state, + verdict=Verdict(ok=False, score=score)) diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..fec25c0 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,73 @@ +"""Tests for the command-line front-end.""" + +import json +import os + +import pytest + +from cyclework import cli + + +def test_version(capsys): + rc = cli.main(["version"]) + assert rc == 0 + assert "cyclework" in capsys.readouterr().out + + +def test_sqrt_solves(capsys): + rc = cli.main(["sqrt", "2", "--quiet"]) + assert rc == 0 + out = capsys.readouterr().out + assert "status = solved" in out + assert "answer = 1.4142" in out + + +def test_sqrt_negative_is_usage_error(capsys): + rc = cli.main(["sqrt", "-1", "--quiet"]) + assert rc == 2 + + +def test_optimize_anneal(capsys): + rc = cli.main(["optimize", "--strategy", "anneal", "--max-iter", "50", "--quiet"]) + assert rc == 0 + assert "answer" in capsys.readouterr().out + + +def test_optimize_restarts(capsys): + rc = cli.main(["optimize", "--strategy", "hill", "--restarts", "3", + "--max-iter", "30", "--quiet"]) + assert rc == 0 + assert "best x=" in capsys.readouterr().out + + +def test_optimize_grid(capsys): + rc = cli.main(["optimize", "--strategy", "grid", "--max-iter", "20", + "--range", "-5", "5", "--quiet"]) + assert rc == 0 + + +def test_demo(capsys): + rc = cli.main(["demo", "--max-iter", "40", "--quiet"]) + assert rc == 0 + assert "metrics" in capsys.readouterr().out + + +def test_jsonl_output(tmp_path, capsys): + p = tmp_path / "trace.jsonl" + rc = cli.main(["sqrt", "9", "--jsonl", str(p), "--quiet"]) + assert rc == 0 + assert os.path.exists(p) + lines = p.read_text(encoding="utf-8").splitlines() + assert len(lines) >= 1 + assert json.loads(lines[0])["index"] == 0 + + +def test_global_option_before_subcommand(capsys): + rc = cli.main(["--quiet", "sqrt", "4"]) + assert rc == 0 + + +def test_target_stop(capsys): + rc = cli.main(["optimize", "--strategy", "anneal", "--target", "0.5", + "--max-iter", "200", "--quiet"]) + assert rc == 0 diff --git a/tests/test_observers.py b/tests/test_observers.py new file mode 100644 index 0000000..f3caf4f --- /dev/null +++ b/tests/test_observers.py @@ -0,0 +1,112 @@ +"""Tests for observers: multi, JsonlWriter, MetricsCollector, ProgressPrinter.""" + +import io +import json +import os + +import pytest + +from cyclework import ( + Engine, Verdict, Iteration, + multi, JsonlWriter, MetricsCollector, ProgressPrinter, +) + + +def _mk_iteration(index, score, ok=False, feedback="", **detail): + return Iteration(index=index, state=index, + verdict=Verdict(ok=ok, score=score, feedback=feedback, detail=detail)) + + +def test_multi_fans_out_and_skips_none(): + a, b = [], [] + obs = multi(a.append, None, b.append) + it = _mk_iteration(0, 1.0) + obs(it) + assert a == [it] and b == [it] + + +def test_jsonl_writer_to_stream(): + buf = io.StringIO() + w = JsonlWriter(buf, include_state=True) + w(_mk_iteration(0, 1.5, feedback="hi", note="x")) + w(_mk_iteration(1, 2.5, ok=True)) + lines = buf.getvalue().strip().splitlines() + assert len(lines) == 2 + rec0 = json.loads(lines[0]) + assert rec0["index"] == 0 and rec0["score"] == 1.5 + assert rec0["feedback"] == "hi" and rec0["detail"]["note"] == "x" + assert rec0["state"] == 0 # int is JSON-native, kept as-is + rec1 = json.loads(lines[1]) + assert rec1["ok"] is True + assert w.count == 2 + + +def test_jsonl_writer_to_path(tmp_path): + p = tmp_path / "trace.jsonl" + w = JsonlWriter.to_path(str(p)) + with w: + w(_mk_iteration(0, 1.0)) + assert os.path.exists(p) + data = [json.loads(l) for l in p.read_text(encoding="utf-8").splitlines()] + assert data[0]["index"] == 0 + assert "state" not in data[0] # include_state defaults False + + +def test_jsonl_writer_engine_integration(tmp_path): + p = tmp_path / "run.jsonl" + w = JsonlWriter.to_path(str(p)) + eng = Engine(lambda x: Verdict(ok=x >= 3, score=float(x)), + lambda s, v: s + 1, max_iterations=10, on_iteration=w) + res = eng.run(0) + w.close() + lines = p.read_text(encoding="utf-8").splitlines() + assert len(lines) == res.iterations + assert json.loads(lines[-1])["ok"] is True + + +def test_metrics_collector_tracks_stats(): + m = MetricsCollector() + for i, s in enumerate([1.0, 3.0, 2.0, 5.0]): + m(_mk_iteration(i, s)) + snap = m.snapshot() + assert snap["count"] == 4 + assert snap["best"] == 5.0 + assert snap["min"] == 1.0 and snap["max"] == 5.0 + assert snap["last"] == 5.0 + assert snap["improvements"] == 3 # 1->3->5 (2.0 not an improvement) + assert abs(m.mean - 2.75) < 1e-9 + + +def test_metrics_collector_ignores_unscored(): + m = MetricsCollector() + m(_mk_iteration(0, None)) + m(_mk_iteration(1, 2.0)) + snap = m.snapshot() + assert snap["count"] == 2 and snap["scored"] == 1 + assert snap["best"] == 2.0 + + +def test_metrics_empty_mean_is_none(): + assert MetricsCollector().mean is None + + +def test_progress_printer_throttles(): + buf = io.StringIO() + pp = ProgressPrinter(every=3, stream=buf) + for i in range(7): + pp(_mk_iteration(i, float(i))) + # prints indices 0,3,6 + lines = buf.getvalue().strip().splitlines() + assert len(lines) == 3 + + +def test_progress_printer_always_prints_ok(): + buf = io.StringIO() + pp = ProgressPrinter(every=100, stream=buf) + pp(_mk_iteration(1, 1.0, ok=True)) # index 1 not a multiple of 100 but ok + assert "OK" in buf.getvalue() + + +def test_progress_printer_bad_every(): + with pytest.raises(ValueError): + ProgressPrinter(every=0) diff --git a/tests/test_policy.py b/tests/test_policy.py new file mode 100644 index 0000000..8766cda --- /dev/null +++ b/tests/test_policy.py @@ -0,0 +1,110 @@ +"""Tests for the retry/backoff policy and its engine integration.""" + +import pytest + +from cyclework import Engine, Verdict, Status, RetryPolicy, NO_RETRY + + +def test_delay_schedule_is_exponential_capped(): + p = RetryPolicy(base_delay=1.0, factor=2.0, max_delay=10.0) + assert p.delay_for(1) == 0.0 # first attempt: no wait + assert p.delay_for(2) == 1.0 + assert p.delay_for(3) == 2.0 + assert p.delay_for(4) == 4.0 + assert p.delay_for(20) == 10.0 # capped + + +def test_should_retry_respects_attempts_and_type(): + p = RetryPolicy(max_attempts=3, retry_on=(ValueError,)) + assert p.should_retry(ValueError("x"), attempt=1) is True + assert p.should_retry(ValueError("x"), attempt=3) is False # no attempts left + assert p.should_retry(KeyError("x"), attempt=1) is False # wrong type + + +def test_run_retries_then_succeeds(): + slept = [] + p = RetryPolicy(max_attempts=3, base_delay=0.1, sleep=slept.append) + state = {"n": 0} + + def flaky(): + state["n"] += 1 + if state["n"] < 3: + raise RuntimeError("transient") + return "ok" + + assert p.run(flaky) == "ok" + assert state["n"] == 3 + assert len(slept) == 2 # slept before attempts 2 and 3 + + +def test_run_reraises_after_exhaustion(): + p = RetryPolicy(max_attempts=2, base_delay=0, sleep=lambda s: None) + + def always(): + raise ValueError("nope") + + with pytest.raises(ValueError): + p.run(always) + + +def test_engine_retries_flaky_check(): + slept = [] + p = RetryPolicy(max_attempts=5, base_delay=0.01, sleep=slept.append) + attempts = {"c": 0} + + def check(x): + attempts["c"] += 1 + if attempts["c"] < 3: + raise ConnectionError("blip") + return Verdict(ok=True, score=1.0) + + eng = Engine(check, lambda s, v: s, max_iterations=3, retry=p) + res = eng.run(0) + assert res.status is Status.SOLVED + assert attempts["c"] == 3 # two failures retried, third succeeds + + +def test_engine_without_retry_aborts_on_first_error(): + def check(x): + raise RuntimeError("boom") + + eng = Engine(check, lambda s, v: s, max_iterations=3) + res = eng.run(0) + assert res.status is Status.ERROR + assert "check failed" in res.error + + +def test_engine_error_after_retries_exhausted(): + p = RetryPolicy(max_attempts=2, base_delay=0, sleep=lambda s: None) + + def check(x): + raise RuntimeError("persistent") + + eng = Engine(check, lambda s, v: s, max_iterations=3, retry=p) + res = eng.run(0) + assert res.status is Status.ERROR + + +def test_no_retry_constant(): + assert NO_RETRY.max_attempts == 1 + + +def test_invalid_policy_params(): + with pytest.raises(ValueError): + RetryPolicy(max_attempts=0) + with pytest.raises(ValueError): + RetryPolicy(factor=0.5) + with pytest.raises(ValueError): + RetryPolicy(jitter=-1) + + +def test_engine_rejects_non_policy_retry(): + with pytest.raises(TypeError): + Engine(lambda x: Verdict(ok=True), lambda s, v: s, retry="fast") + + +def test_jitter_bounded(): + import random + p = RetryPolicy(base_delay=1.0, factor=1.0, jitter=0.5, rng=random.Random(0)) + d = p.delay_for(2) + assert 1.0 <= d <= 1.5 diff --git a/tests/test_search.py b/tests/test_search.py new file mode 100644 index 0000000..4ff59f6 --- /dev/null +++ b/tests/test_search.py @@ -0,0 +1,106 @@ +"""Tests for keep-best-of-N and parallel restart orchestration.""" + +import math + +import pytest + +from cyclework import Engine, Verdict, Status, best_of, parallel_best_of, BestOf + + +def _bumpy_engine(seed_strategy=None, max_iterations=100): + import cyclework.strategies as strat + + def f(x): + return math.sin(3 * x) - 0.1 * (x - 2) ** 2 + + def check(x): + return Verdict(ok=False, score=f(x)) + + revise = strat.simulated_anneal(step=0.3, seed=seed_strategy or 0) + return Engine(check, revise, max_iterations=max_iterations) + + +def test_best_of_returns_best_run(): + def check(x): + return Verdict(ok=False, score=-(x - 5) ** 2) # best when start near 5 + + eng = Engine(check, lambda s, v: s, max_iterations=1) # 1 iter: score the seed + bundle = best_of(eng, seeds=[0.0, 5.0, 9.0], stop_when_solved=False) + assert isinstance(bundle, BestOf) + assert len(bundle.runs) == 3 + assert bundle.best.best.state == 5.0 + assert bundle.best_index == 1 + + +def test_best_of_stops_early_on_solve(): + def check(x): + return Verdict(ok=(x == 2), score=float(x)) + + eng = Engine(check, lambda s, v: s, max_iterations=1) + bundle = best_of(eng, seeds=[1.0, 2.0, 3.0], stop_when_solved=True) + # stops after the solved run (seed 2.0), never runs seed 3.0 + assert len(bundle.runs) == 2 + assert bundle.solved + + +def test_best_of_summary(): + def check(x): + return Verdict(ok=False, score=float(x)) + eng = Engine(check, lambda s, v: s, max_iterations=1) + bundle = best_of(eng, seeds=[1.0, 4.0, 2.0], stop_when_solved=False) + s = bundle.summary() + assert s["runs"] == 3 + assert s["best_score"] == 4.0 + assert s["score_spread"] == 3.0 # 4 - 1 + + +def test_best_of_empty_seeds_rejected(): + eng = Engine(lambda x: Verdict(ok=True), lambda s, v: s) + with pytest.raises(ValueError): + best_of(eng, seeds=[]) + + +def test_best_of_accepts_factory_for_stateful_engine(): + import cyclework.strategies as strat + + def factory(): + def check(x): + return Verdict(ok=False, score=-(x - 3) ** 2) + return Engine(check, strat.hill_climb(step=0.4), max_iterations=40) + + bundle = best_of(factory, seeds=[0.0, 6.0], stop_when_solved=False) + assert len(bundle.runs) == 2 + # both restarts should climb toward the peak at x=3 + assert abs(bundle.best.best.state - 3) < 0.5 + + +def test_parallel_best_of_matches_serial(): + def factory(): + def check(x): + return Verdict(ok=False, score=-(x - 5) ** 2) + return Engine(check, lambda s, v: s, max_iterations=1) + + bundle = parallel_best_of(factory, seeds=[0.0, 5.0, 9.0], max_workers=3) + assert bundle.best.best.state == 5.0 + assert len(bundle.runs) == 3 + + +def test_parallel_best_of_fresh_engine_per_run(): + # each factory() call gets its own stateful strategy; verify no crash and + # all runs complete + import cyclework.strategies as strat + + def factory(): + def check(x): + return Verdict(ok=False, score=-(x - 3) ** 2) + return Engine(check, strat.hill_climb(step=0.5), max_iterations=30) + + bundle = parallel_best_of(factory, seeds=[0.0, 6.0], max_workers=2) + assert len(bundle.runs) == 2 + assert all(r.status is Status.EXHAUSTED for r in bundle.runs) + + +def test_parallel_best_of_empty_rejected(): + with pytest.raises(ValueError): + parallel_best_of(lambda: Engine(lambda x: Verdict(ok=True), lambda s, v: s), + seeds=[]) diff --git a/tests/test_stopping.py b/tests/test_stopping.py new file mode 100644 index 0000000..5e11cfe --- /dev/null +++ b/tests/test_stopping.py @@ -0,0 +1,180 @@ +"""Tests for pluggable stop conditions and their integration with the engine.""" + +import itertools + +import pytest + +from cyclework import ( + Engine, Verdict, Status, + MaxSeconds, TargetScore, NoImprovement, ScoreThreshold, Predicate, + StopCondition, as_stop_condition, +) + + +def _counter_engine(scores, **kw): + """Engine whose check returns the next scripted score each call.""" + it = iter(scores) + + def check(x): + return Verdict(ok=False, score=next(it)) + + def revise(x, v): + return x + 1 + + return Engine(check, revise, max_iterations=len(scores), **kw) + + +# ---- TargetScore --------------------------------------------------------- + +def test_target_score_stops_with_solved(): + eng = _counter_engine([0.1, 0.5, 0.9, 1.0], stop=TargetScore(0.8)) + res = eng.run(0) + assert res.status is Status.SOLVED + assert res.iterations == 3 # 0.9 crosses 0.8 + + +def test_target_score_ignores_unscored(): + def check(x): + return Verdict(ok=False, score=None) + eng = Engine(check, lambda s, v: s, max_iterations=3, stop=TargetScore(0.0)) + res = eng.run(0) + assert res.status is Status.EXHAUSTED + + +# ---- MaxSeconds ---------------------------------------------------------- + +def test_max_seconds_fires_with_fake_clock(): + ticks = itertools.count(0.0, 1.0) # 0,1,2,3,... seconds per read + clock = lambda: next(ticks) + # reset() reads clock once (start), then each eval reads once + stop = MaxSeconds(2.5, clock=clock) + eng = _counter_engine([0.0] * 10, stop=stop) + res = eng.run(0) + assert res.status is Status.TIMEOUT + + +def test_max_seconds_zero_stops_immediately(): + stop = MaxSeconds(0.0) + eng = _counter_engine([0.0, 0.0, 0.0], stop=stop) + res = eng.run(0) + assert res.status is Status.TIMEOUT + assert res.iterations == 1 + + +def test_max_seconds_negative_rejected(): + with pytest.raises(ValueError): + MaxSeconds(-1) + + +# ---- NoImprovement ------------------------------------------------------- + +def test_no_improvement_plateaus(): + stop = NoImprovement(patience=2) + eng = _counter_engine([1.0, 1.0, 1.0, 1.0], stop=stop) + res = eng.run(0) + assert res.status is Status.PLATEAU + + +def test_no_improvement_resets_on_gain(): + stop = NoImprovement(patience=2) + # improves each step -> never plateaus, runs out + eng = _counter_engine([1.0, 2.0, 3.0], stop=stop) + res = eng.run(0) + assert res.status is Status.EXHAUSTED + + +def test_no_improvement_relative_delta(): + # 100 -> 100.5 is <0.5% gain, counts as stale under rel_delta=0.01 + stop = NoImprovement(patience=1, rel_delta=0.01) + eng = _counter_engine([100.0, 100.5, 100.6], stop=stop) + res = eng.run(0) + assert res.status is Status.PLATEAU + + +def test_no_improvement_bad_patience(): + with pytest.raises(ValueError): + NoImprovement(patience=0) + + +# ---- ScoreThreshold (divergence guard) ----------------------------------- + +def test_score_threshold_aborts_on_divergence(): + stop = ScoreThreshold(floor=-10.0) + eng = _counter_engine([-1.0, -5.0, -50.0, -1.0], stop=stop) + res = eng.run(0) + assert res.status is Status.ABORTED + assert res.iterations == 3 + + +# ---- Predicate ----------------------------------------------------------- + +def test_predicate_on_state(): + def check(x): + return Verdict(ok=False, score=float(x)) + stop = Predicate(lambda s: s >= 3) + eng = Engine(check, lambda s, v: s + 1, max_iterations=10, stop=stop) + res = eng.run(0) + assert res.status is Status.SOLVED + assert res.state == 3 + + +# ---- composition --------------------------------------------------------- + +def test_or_composition_first_wins(): + combined = TargetScore(100.0) | ScoreThreshold(-5.0) + eng = _counter_engine([-1.0, -10.0], stop=combined) + res = eng.run(0) + assert res.status is Status.ABORTED # floor hit before target + + +def test_or_flattens_and_names(): + combined = TargetScore(1.0) | NoImprovement(2) | MaxSeconds(1) + assert isinstance(combined, StopCondition) + assert "target_score" in combined.name and "max_seconds" in combined.name + + +def test_ok_verdict_beats_stop_condition(): + # a passing verdict must win even if a stop condition would also fire + def check(x): + return Verdict(ok=True, score=-100.0) + eng = Engine(check, lambda s, v: s, max_iterations=5, + stop=ScoreThreshold(0.0)) + res = eng.run(0) + assert res.status is Status.SOLVED + + +def test_as_stop_condition_wraps_callable(): + calls = [] + + def raw(it, hist): + calls.append(it.index) + return Status.ABORTED if it.index == 1 else None + + sc = as_stop_condition(raw) + eng = _counter_engine([0.0, 0.0, 0.0], stop=sc) + res = eng.run(0) + assert res.status is Status.ABORTED + assert calls == [0, 1] + + +def test_as_stop_condition_rejects_noncallable(): + with pytest.raises(TypeError): + as_stop_condition(42) + + +def test_stop_condition_is_reset_across_runs(): + stop = NoImprovement(patience=1) + eng = _counter_engine([1.0, 1.0], stop=stop) + r1 = eng.run(0) + assert r1.status is Status.PLATEAU + # a second run must not carry stale state; fresh improving sequence + eng2 = _counter_engine([5.0, 6.0], stop=stop) + r2 = eng2.run(0) + assert r2.status is Status.EXHAUSTED + + +def test_stop_works_in_streaming_cycle(): + eng = _counter_engine([0.1, 0.2, 0.9, 1.0], stop=TargetScore(0.8)) + seen = list(eng.cycle(0)) + # cycle stops after the iteration whose score crosses target + assert len(seen) == 3 diff --git a/tests/test_strategies.py b/tests/test_strategies.py new file mode 100644 index 0000000..ea2c013 --- /dev/null +++ b/tests/test_strategies.py @@ -0,0 +1,114 @@ +"""Tests for the revise strategy library.""" + +import math + +import pytest + +from cyclework import Engine, Verdict, strategies + + +def test_hill_climb_finds_scalar_max(): + # maximize -(x-3)^2, peak at x=3 + def check(x): + return Verdict(ok=abs(x - 3) < 1e-3, score=-(x - 3) ** 2) + + eng = Engine(check, strategies.hill_climb(step=1.0, shrink=0.5), + max_iterations=200) + res = eng.run(0.0) + assert res.best is not None + assert abs(res.best.state - 3) < 0.1 + + +def test_hill_climb_requires_score(): + revise = strategies.hill_climb() + revise(0.0, Verdict(ok=False, score=1.0)) # priming call ok + with pytest.raises(ValueError): + revise(0.0, Verdict(ok=False, score=None)) + + +def test_simulated_anneal_escapes_local_optimum(): + # bumpy function with a local and a global max; global near x=2 + def f(x): + return math.sin(3 * x) - 0.1 * (x - 2) ** 2 + + def check(x): + return Verdict(ok=False, score=f(x)) + + eng = Engine(check, strategies.simulated_anneal(step=0.3, seed=1), + max_iterations=300) + res = eng.run(-4.0) + assert res.best is not None + # should get meaningfully above a poor start + assert res.best.score > f(-4.0) + 0.5 + + +def test_simulated_anneal_requires_score(): + revise = strategies.simulated_anneal() + with pytest.raises(ValueError): + revise(0.0, Verdict(ok=False, score=None)) + + +def test_simulated_anneal_reproducible(): + def check(x): + return Verdict(ok=False, score=-(x - 1) ** 2) + r1 = Engine(check, strategies.simulated_anneal(step=0.2, seed=7), + max_iterations=50).run(0.0) + r2 = Engine(check, strategies.simulated_anneal(step=0.2, seed=7), + max_iterations=50).run(0.0) + assert r1.best.score == r2.best.score + + +def test_grid_step_sweeps_values(): + vals = [0.0, 1.0, 2.0, 3.0, 4.0] + + def check(x): + return Verdict(ok=False, score=-(x - 3) ** 2) # best at x=3 + + eng = Engine(check, strategies.grid_step(vals), max_iterations=len(vals)) + res = eng.run(vals[0]) + assert res.best.state == 3.0 + + +def test_grid_step_holds_last_when_exhausted(): + revise = strategies.grid_step([10.0, 20.0]) + v = Verdict(ok=False, score=0.0) + assert revise(10.0, v) == 20.0 + assert revise(20.0, v) == 20.0 # holds last + assert revise(20.0, v) == 20.0 + + +def test_grid_step_empty_rejected(): + with pytest.raises(ValueError): + strategies.grid_step([]) + + +def test_generic_hill_climb_on_tuple_state(): + # maximize sum of a small int vector, neighbors flip one coord +/-1, + # bounded to [0, 3] + target_sum = 9 # 3+3+3 + + def neighbors(state): + out = [] + for i in range(len(state)): + for d in (-1, 1): + nv = list(state) + nv[i] = min(3, max(0, nv[i] + d)) + out.append(tuple(nv)) + return out + + def score_of(state): + return float(sum(state)) + + def check(state): + return Verdict(ok=sum(state) >= target_sum, score=score_of(state)) + + revise = strategies.generic_hill_climb(neighbors, score_of, seed=0) + eng = Engine(check, revise, max_iterations=50) + res = eng.run((0, 0, 0)) + assert res.solved + assert res.state == (3, 3, 3) + + +def test_generic_hill_climb_no_neighbors_holds(): + revise = strategies.generic_hill_climb(lambda s: [], lambda s: 0.0) + assert revise("x", Verdict(ok=False, score=0.0)) == "x" diff --git a/tests/test_trace_extras.py b/tests/test_trace_extras.py new file mode 100644 index 0000000..da78856 --- /dev/null +++ b/tests/test_trace_extras.py @@ -0,0 +1,64 @@ +"""Tests for the enriched Result.metrics / to_dict and new Status members.""" + +from cyclework import Engine, Verdict, Status + + +def _run_scores(scores): + it = iter(scores) + eng = Engine(lambda x: Verdict(ok=False, score=next(it)), + lambda s, v: s + 1, max_iterations=len(scores)) + return eng.run(0) + + +def test_metrics_aggregates(): + res = _run_scores([1.0, 3.0, 2.0, 5.0]) + m = res.metrics() + assert m["scored_iterations"] == 4 + assert m["min_score"] == 1.0 + assert m["max_score"] == 5.0 + assert abs(m["mean_score"] - 2.75) < 1e-9 + assert m["best_index"] == 3 + assert m["total_improvement"] == 4.0 # best(5) - first(1) + + +def test_metrics_no_scores(): + eng = Engine(lambda x: Verdict(ok=False, score=None), + lambda s, v: s + 1, max_iterations=3) + m = eng.run(0).metrics() + assert m["scored_iterations"] == 0 + assert m["min_score"] is None + assert m["mean_score"] is None + assert m["total_improvement"] is None + + +def test_metrics_superset_of_summary(): + res = _run_scores([1.0, 2.0]) + m, s = res.metrics(), res.summary() + for k in s: + assert m[k] == s[k] + + +def test_to_dict_shape(): + res = _run_scores([1.0, 2.0, 3.0]) + d = res.to_dict() + assert set(d) == {"metrics", "history"} + assert len(d["history"]) == 3 + assert d["history"][0] == {"index": 0, "ok": False, "score": 1.0, "feedback": ""} + + +def test_to_dict_include_states(): + res = _run_scores([1.0, 2.0]) + d = res.to_dict(include_states=True) + assert d["history"][0]["state"] == "0" + + +def test_new_status_members_exist(): + assert Status.TIMEOUT.value == "timeout" + assert Status.ABORTED.value == "aborted" + assert Status.TIMEOUT.terminal is True + + +def test_new_statuses_not_solved(): + res = _run_scores([1.0]) + res.status = Status.TIMEOUT + assert res.solved is False