diff --git a/README.md b/README.md index 195c50d..ec68f9e 100644 --- a/README.md +++ b/README.md @@ -14,7 +14,7 @@ Ask yourself: A lot of useful work has that one shape: produce a candidate, check it against a goal, use the feedback to produce a better one — until it's good enough, it stops improving, or you run out of budget. Solvers, optimizers, retry-with-feedback agent loops, and self-correcting generators are all *this loop*. -`cyclework` makes it a first-class, inspectable object instead of a `while` buried in a function. You supply two functions — how to **check** a candidate and how to **revise** it given feedback — and the engine runs the cycle, detects convergence and plateaus, enforces a budget, captures errors, and hands back a full trace. +`cyclework` makes it a first-class, inspectable object instead of a `while` buried in a function. You supply two functions — how to **check** a candidate and how to **revise** it given feedback — and the engine runs the cycle, detects convergence and plateaus, enforces a budget, captures errors (including in the optional `seed`), validates its own configuration, and hands back a full trace. - **Domain-agnostic.** The candidate ("state") can be a number, a string, a dict, an AST — anything. - **Inspectable.** Every iteration is recorded; you get the answer *and* the path to it. @@ -108,7 +108,7 @@ model, stopping conditions, and why the loop is a first-class object). - **Solved** — a verdict comes back `ok=True`. - **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`, or `seed` raised (or `check` returned something other than a `Verdict`); the run aborts cleanly with a message naming the stage and iteration, and the trace + best candidate captured so far. ### Observability @@ -120,12 +120,12 @@ Engine(check, revise, on_iteration=lambda it: print(it.index, it.score)) ## Demos -Five 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). +Twenty runnable, narrated scenarios in [`demos/`](demos/), each for a different +audience or facet of the engine — 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 twenty, end to end python demos/02_numeric_solvers.py # or just one ``` @@ -136,6 +136,7 @@ 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–20 | [see `docs/DEMOS.md`](docs/DEMOS.md) | resilience, ML, data-cleaning, compliance, reliability, numerical … | Error recovery, gradient descent, constraint repair, retry-with-backoff, audit logging, best-of non-convergent search, Newton vs bisection, and more. | On a cp1252 (Windows) console, prefix with `PYTHONUTF8=1`. @@ -143,7 +144,7 @@ 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 # 136 tests (engine, data model, examples, and demo smoke tests) ``` ## License diff --git a/cyclework/engine.py b/cyclework/engine.py index 80c1f02..ff9e7a4 100644 --- a/cyclework/engine.py +++ b/cyclework/engine.py @@ -11,8 +11,11 @@ stopping when a verdict is `ok`, when the score plateaus (no improvement of at least `min_delta` for `patience` checks), or when `max_iterations` is reached. -A stage raising an exception aborts cleanly with `Status.ERROR` and the trace -collected so far. +Any stage (`seed`, `check`, `revise`) that raises — or a `check` that returns +something other than a `Verdict` — aborts cleanly with `Status.ERROR`, a message +naming the stage and iteration, and the trace collected so far. Constructor +arguments are validated up front (callable stages; non-negative budget/min_delta; +patience >= 1). Convention: `score` is higher-is-better. For a "lower is better" objective (an error/cost), return its negation as the score. @@ -47,6 +50,16 @@ def __init__( raise ValueError("max_iterations must be >= 0") if patience is not None and patience < 1: raise ValueError("patience must be >= 1") + if min_delta < 0: + raise ValueError("min_delta must be >= 0") + if not callable(check): + raise TypeError("check must be callable: check(state) -> Verdict") + if not callable(revise): + raise TypeError("revise must be callable: revise(state, verdict) -> state") + if seed is not None and not callable(seed): + raise TypeError("seed must be callable: seed(initial) -> state") + if on_iteration is not None and not callable(on_iteration): + raise TypeError("on_iteration must be callable: on_iteration(iteration) -> None") self.check = check self.revise = revise self.max_iterations = max_iterations @@ -57,9 +70,16 @@ def __init__( # ---- streaming form: yields each iteration as it happens ------------- def cycle(self, initial: Any) -> Iterator[Iteration]: + """Yield each `Iteration` as it happens. + + Unlike `run()`, the streaming form does not trap stage exceptions or + detect plateaus — the caller owns the loop and can stop early or wrap + it as it sees fit. A `check` that returns something other than a + `Verdict` still raises a clear `TypeError`. + """ state = self.seed(initial) if self.seed else initial for i in range(self.max_iterations): - verdict = self.check(state) + verdict = _as_verdict(self.check(state), "check", i) it = Iteration(index=i, state=state, verdict=verdict) if self.on_iteration: self.on_iteration(it) @@ -75,11 +95,21 @@ def run(self, initial: Any) -> Result: status = Status.EXHAUSTED last_improved: Optional[float] = None stale = 0 - state = self.seed(initial) if self.seed else initial + + # the seed is a stage too: surface its failures as ERROR, just like + # check/revise, rather than letting them escape the engine. + if self.seed is not None: + try: + state = self.seed(initial) + except Exception as e: # noqa: BLE001 - surface seed errors as ERROR status + return Result(Status.ERROR, initial, 0, history, best, + error=f"seed failed: {e}") + else: + state = initial for i in range(self.max_iterations): try: - verdict = self.check(state) + verdict = _as_verdict(self.check(state), "check", i) 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}") @@ -114,6 +144,23 @@ def run(self, initial: Any) -> Result: return Result(status, state, len(history), history, best) +def _as_verdict(value: Any, stage: str, index: int) -> Verdict: + """Guard that a stage returned a `Verdict`. + + Returning a bare bool/tuple/None from `check` is the single most common + misuse; without this guard it surfaces deep in the engine as a cryptic + `AttributeError`. Here it becomes an actionable `TypeError` naming the + stage and iteration. + """ + if not isinstance(value, Verdict): + raise TypeError( + f"{stage} must return a Verdict at iteration {index}, " + f"got {type(value).__name__}: {value!r}. " + f"Use Verdict(ok=...), Verdict.passed(...), or Verdict.failed(...)." + ) + return value + + def _better(current: Optional[Iteration], candidate: Iteration) -> Iteration: """Keep the higher-scoring iteration; unscored candidates never displace a scored best, but become best if nothing better exists yet.""" diff --git a/demos/06_error_recovery.py b/demos/06_error_recovery.py new file mode 100644 index 0000000..9647b98 --- /dev/null +++ b/demos/06_error_recovery.py @@ -0,0 +1,64 @@ +"""Scenario 6 - resilience engineers: failures are first-class outcomes. + +Real refinement stages fail: a model endpoint times out, a `revise` hits a +divide-by-zero, a `seed` preprocessor chokes on bad input. cyclework does not +let those escape as raw tracebacks - it surfaces them as `Status.ERROR` with a +message naming the stage and iteration, and it preserves the trace and the best +candidate seen *before* the failure, so the caller can still recover something. + +This demo triggers a failure in each of the three stages (check, revise, seed) +and shows the engine naming the stage, keeping the partial history, and handing +back the best pre-failure candidate. +""" +from _common import Engine, Verdict, rule, show_summary, show_trace + + +def main() -> None: + rule("ERROR RECOVERY - stage failures become inspectable outcomes") + + # --- a check that explodes part-way through --------------------------- + print("\n1) check() raises on the 3rd call - history before it survives:\n") + calls = {"n": 0} + + def flaky_check(s): + calls["n"] += 1 + if calls["n"] == 3: + raise RuntimeError("scoring service 503") + return Verdict(ok=False, score=float(s)) + + res = Engine(flaky_check, lambda s, v: s + 1, max_iterations=10).run(0) + show_trace(res, fmt_state=str) + show_summary(res) + print(f" -> {len(res.history)} good iterations kept; best score " + f"{res.best.score} from before the failure.") + + # --- a revise that divides by zero ------------------------------------ + print("\n2) revise() raises - the engine names the stage and iteration:\n") + + def bad_revise(s, v): + return 1 / 0 # noqa: B018 + + res = Engine(lambda s: Verdict(ok=False, score=1.0), bad_revise, + max_iterations=5).run(10) + show_summary(res) + + # --- a seed that fails before the loop even starts -------------------- + print("\n3) seed() raises before any iteration - clean ERROR, empty trace:\n") + + def bad_seed(_x): + raise ValueError("could not parse initial input") + + res = Engine(lambda s: Verdict(True), lambda s, v: s, + seed=bad_seed).run("garbage") + show_summary(res) + print(f" -> history length {len(res.history)} (nothing ran), " + f"status named the failure instead of crashing the caller.") + + print( + "\nEvery stage failure is a value you can branch on (res.status, " + "res.error), not an exception that takes down the whole pipeline." + ) + + +if __name__ == "__main__": + main() diff --git a/demos/07_gradient_descent.py b/demos/07_gradient_descent.py new file mode 100644 index 0000000..a575e55 --- /dev/null +++ b/demos/07_gradient_descent.py @@ -0,0 +1,58 @@ +"""Scenario 7 - ML / optimization engineers: descent as a refinement loop. + +Gradient descent is the archetypal "propose, check, revise" loop: the check +scores how low the loss is, the revise steps downhill. Because cyclework tracks +the best candidate seen, a descent that overshoots and bounces still hands back +its lowest-loss point - not wherever it happened to stop. + +This demo minimizes a 1-D quadratic with three learning rates - one that +converges smoothly, one that converges slowly (and exhausts its budget), and +one so large it oscillates - and shows the engine naming each outcome while +always recovering the best point. +""" +from _common import Engine, Status, Verdict, rule, show_summary + + +def descend(lr: float, max_iterations: int = 60): + """Minimize f(x) = (x - 3)^2. score = -loss (higher is better).""" + target = 3.0 + + def check(x): + loss = (x - target) ** 2 + return Verdict(ok=loss <= 1e-9, score=-loss, detail={"loss": loss}) + + def revise(x, _v): + grad = 2 * (x - target) # df/dx + return x - lr * grad + + return Engine(check, revise, max_iterations=max_iterations).run(0.0) + + +def main() -> None: + rule("GRADIENT DESCENT - minimize a quadratic, keep the best point") + print("\nMinimize f(x) = (x - 3)^2 from x0 = 0, varying the learning rate.\n") + + for lr, label in [(0.3, "well-tuned"), (0.02, "too small"), (1.05, "too large")]: + res = descend(lr) + best_x = res.best.state + best_loss = res.best.verdict.detail["loss"] + print(f"lr={lr:<5} ({label}):") + print(f" status={res.status.value:<9} iterations={res.iterations:<3} " + f"best_x={best_x:.6f} best_loss={best_loss:.3e}") + if res.status is Status.SOLVED: + print(" -> converged to the minimum.") + elif res.status is Status.EXHAUSTED: + print(" -> ran out of budget; best point so far is still usable.") + print() + + print("Verbose summary of the well-tuned run:") + show_summary(descend(0.3)) + + print( + "\nSame engine, three regimes. The loop never loses the best candidate, " + "so even a non-converging run returns something you can use." + ) + + +if __name__ == "__main__": + main() diff --git a/demos/08_text_normalizer.py b/demos/08_text_normalizer.py new file mode 100644 index 0000000..4ae9b70 --- /dev/null +++ b/demos/08_text_normalizer.py @@ -0,0 +1,80 @@ +"""Scenario 8 - data-cleaning engineers: a multi-rule normalizer. + +The bundled slug refiner fixes one rule per cycle; here we build a richer +normalizer for free-text product codes from the same primitives - each cycle +the check reports the highest-priority violation and revise applies exactly +that fix. The point: a "clean this messy field" pipeline is a refinement loop, +and the trace tells you *which* rules fired, in what order, for any input. +""" +import re + +from _common import Engine, Verdict, rule, show_summary, show_trace + + +def normalize_code(raw: str): + """Normalize a product code: uppercase, A-Z0-9 and single dashes, <= 16.""" + MAX = 16 + + def check(s: str) -> Verdict: + n = sum(c not in "-" for c in s) + score = float(n) # prefer candidates closer to a valid form + if s != s.strip(): + return Verdict.failed("has surrounding whitespace", score=score, fix="trim") + if s != s.upper(): + return Verdict.failed("not uppercase", score=score, fix="upper") + if re.search(r"\s", s): + return Verdict.failed("internal whitespace", score=score, fix="space") + if re.search(r"[^A-Z0-9-]", s): + return Verdict.failed("illegal characters", score=score, fix="strip") + if "--" in s or s.startswith("-") or s.endswith("-"): + return Verdict.failed("dash noise", score=score, fix="dash") + if len(s) > MAX: + return Verdict.failed("too long", score=score, fix="trunc") + return Verdict.passed(score=score) + + def revise(s: str, v: Verdict) -> str: + fix = v.detail.get("fix") + if fix == "trim": + return s.strip() + if fix == "upper": + return s.upper() + if fix == "space": + return re.sub(r"\s+", "-", s) + if fix == "strip": + return re.sub(r"[^A-Z0-9-]", "", s) + if fix == "dash": + return re.sub(r"-+", "-", s).strip("-") + if fix == "trunc": + return s[:MAX].rstrip("-") + return s + + return Engine(check, revise, max_iterations=25).run(raw) + + +CASES = [ + " sku 12 / abc!! ", + "Widget---Pro---2026", + "VALID-CODE-01", + "this-is-a-really-long-product-code-that-overflows", +] + + +def main() -> None: + rule("TEXT NORMALIZER - multi-rule cleaning, one fix per cycle") + print("\nNormalize free-text product codes to UPPER / [A-Z0-9-] / <=16 chars.") + print("Each cycle names the top violation; the trace shows which rules fired.\n") + + for raw in CASES: + print(f"input: {raw!r}") + res = normalize_code(raw) + show_trace(res, fmt_state=repr) + print(f" => {res.state!r}") + show_summary(res) + print() + + print("The same propose/check/revise shape scales to any rule set you can " + "rank by priority.") + + +if __name__ == "__main__": + main() diff --git a/demos/09_misuse_guards.py b/demos/09_misuse_guards.py new file mode 100644 index 0000000..8346041 --- /dev/null +++ b/demos/09_misuse_guards.py @@ -0,0 +1,51 @@ +"""Scenario 9 - library users: the engine fails loudly, not cryptically. + +The most common way to misuse a propose/check/revise engine is to have `check` +return a bare bool (or None, or a tuple) instead of a `Verdict`. Without a +guard that surfaces deep inside the loop as a baffling AttributeError. cyclework +catches it at the boundary and tells you exactly what went wrong, where. + +This demo deliberately misconfigures the engine in several ways and shows the +clear error each one produces - config validation up front (TypeError / +ValueError) and a wrong return type surfaced as a named ERROR result. +""" +from _common import Engine, Verdict, rule + + +def main() -> None: + rule("MISUSE GUARDS - clear errors beat cryptic tracebacks") + + print("\n1) check returns a bare bool instead of a Verdict:\n") + res = Engine(lambda s: s >= 3, lambda s, v: s + 1, max_iterations=5).run(0) + print(f" status={res.status.value}") + print(f" error: {res.error}") + + print("\n2) check returns None (forgot to return at all):\n") + res = Engine(lambda s: None, lambda s, v: s).run(0) + print(f" status={res.status.value}") + print(f" error: {res.error}") + + print("\n3) constructor validation - caught immediately, before any run:\n") + for label, thunk in [ + ("max_iterations=-1", lambda: Engine(lambda s: Verdict(True), lambda s, v: s, max_iterations=-1)), + ("patience=0", lambda: Engine(lambda s: Verdict(True), lambda s, v: s, patience=0)), + ("min_delta=-0.5", lambda: Engine(lambda s: Verdict(True), lambda s, v: s, min_delta=-0.5)), + ("check not callable", lambda: Engine(123, lambda s, v: s)), + ("revise not callable", lambda: Engine(lambda s: Verdict(True), "nope")), + ("seed not callable", lambda: Engine(lambda s: Verdict(True), lambda s, v: s, seed=7)), + ]: + try: + thunk() + print(f" {label:<22} -> (no error - unexpected!)") + except (ValueError, TypeError) as e: + print(f" {label:<22} -> {type(e).__name__}: {e}") + + print( + "\nMisconfiguration is caught at construction; a wrong return type is " + "surfaced as a named ERROR result you can branch on - never a surprise " + "AttributeError ten frames deep." + ) + + +if __name__ == "__main__": + main() diff --git a/demos/10_collatz_and_budget.py b/demos/10_collatz_and_budget.py new file mode 100644 index 0000000..bbb80c6 --- /dev/null +++ b/demos/10_collatz_and_budget.py @@ -0,0 +1,52 @@ +"""Scenario 10 - algorithm explorers: bounded iteration with a hard budget. + +Some loops are not guaranteed to terminate (or you simply do not want them to +run forever). The Collatz sequence is the classic example: iterate +n -> n/2 (even) or 3n+1 (odd) and you *conjecture* it reaches 1, but you cannot +prove it for every start. cyclework turns that into a safe, bounded experiment: +set a budget, run, and read back whether it solved or exhausted - never a hang. + +This demo runs Collatz from several starts under a fixed budget and reports the +stopping time (number of steps to reach 1) or EXHAUSTED if the budget ran out. +""" +from _common import Engine, Status, Verdict, rule + + +def collatz_steps(start: int, max_iterations: int = 1000): + def check(n: int) -> Verdict: + # score = -n so 'closer to 1' reads as higher; ok when we hit 1 + return Verdict(ok=n == 1, score=-float(n)) + + def revise(n: int, _v: Verdict) -> int: + return n // 2 if n % 2 == 0 else 3 * n + 1 + + return Engine(check, revise, max_iterations=max_iterations).run(start) + + +def main() -> None: + rule("COLLATZ & BUDGET - unprovable loops made safe by a hard cap") + print("\nIterate n -> n/2 (even) / 3n+1 (odd) until it reaches 1.") + print("A budget guarantees termination even where the math does not.\n") + + for start in [1, 6, 27, 97, 871]: + res = collatz_steps(start, max_iterations=500) + steps = res.iterations - 1 # checks include the starting n + if res.status is Status.SOLVED: + print(f" start={start:<5} reached 1 in {steps:>3} steps") + else: + print(f" start={start:<5} {res.status.value} after {steps} steps " + f"(min value seen = {int(-res.best.score)})") + + print("\nNow with a deliberately tiny budget so it cannot finish:") + res = collatz_steps(27, max_iterations=10) + print(f" start=27 budget=10 -> status={res.status.value}, " + f"reached n={res.state} (not yet 1).") + + print( + "\nThe budget is the safety rail: the loop always returns, and the " + "status tells you whether it got there or simply ran out of room." + ) + + +if __name__ == "__main__": + main() diff --git a/demos/11_seed_preprocessing.py b/demos/11_seed_preprocessing.py new file mode 100644 index 0000000..eb48749 --- /dev/null +++ b/demos/11_seed_preprocessing.py @@ -0,0 +1,51 @@ +"""Scenario 11 - pipeline authors: the seed normalizes input before the loop. + +A refinement loop often wants its starting candidate in a canonical form first: +parse a string into a number, clamp into a valid range, decompress, validate. +cyclework's optional `seed(initial) -> state` runs exactly once, before the +first check - and (unlike a bare preprocess) a failing seed is surfaced as a +clean `Status.ERROR`, not a stray exception. + +This demo uses a seed to parse and clamp messy user input into a number the +loop can refine, and also shows a seed that rejects bad input cleanly. +""" +from _common import Engine, Status, Verdict, rule, show_summary + + +def parse_and_clamp(raw: str): + """Seed turns a string into a float clamped to [0, 100]; loop counts to 100.""" + + def seed(s: str) -> float: + val = float(s.strip()) # raises ValueError on garbage -> ERROR + return max(0.0, min(100.0, val)) + + def check(x: float) -> Verdict: + return Verdict(ok=x >= 100.0, score=x) + + def revise(x: float, _v: Verdict) -> float: + return x + 25.0 + + return Engine(check, revise, seed=seed, max_iterations=20).run(raw) + + +def main() -> None: + rule("SEED PREPROCESSING - canonicalize input once, before the loop") + print("\nSeed parses a string and clamps it to [0,100]; the loop climbs to 100.\n") + + for raw in [" 10 ", "-50", "200", "62.5"]: + res = parse_and_clamp(raw) + print(f" input={raw!r:<8} seeded+ran -> state={res.state} " + f"status={res.status.value} iterations={res.iterations}") + + print("\nA seed that cannot parse its input fails as a clean ERROR:\n") + res = parse_and_clamp("not-a-number") + show_summary(res) + + print( + "\nThe seed is the loop's front door: one-time normalization, with the " + "same honest error handling as every other stage." + ) + + +if __name__ == "__main__": + main() diff --git a/demos/12_plateau_early_stop.py b/demos/12_plateau_early_stop.py new file mode 100644 index 0000000..f0cbc5d --- /dev/null +++ b/demos/12_plateau_early_stop.py @@ -0,0 +1,56 @@ +"""Scenario 12 - training-loop owners: stop early when learning stalls. + +You rarely want to burn the full budget once a loop has stopped improving. +With `patience` and `min_delta`, cyclework detects a plateau - N consecutive +checks that fail to improve the score by at least min_delta - and stops, handing +back the best candidate it saw. This is the same early-stopping you would wire +into a training loop, but as a property of the engine. + +This demo models a noisy learning curve that climbs, then flattens, and shows +how patience/min_delta change *when* the engine calls it quits - and that the +best epoch is always retained regardless of where it stopped. +""" +from _common import Engine, Status, Verdict, rule + + +# a hand-built "learning curve": fast early gains, then it flattens out. +CURVE = [0.10, 0.45, 0.70, 0.82, 0.88, 0.905, 0.915, 0.920, 0.922, 0.923, + 0.9235, 0.9237, 0.9238, 0.92385, 0.92386] + + +def run(patience: int, min_delta: float): + scores = iter(CURVE) + + def check(_s): + try: + return Verdict(ok=False, score=next(scores)) + except StopIteration: + return Verdict(ok=False, score=CURVE[-1]) + + return Engine(check, lambda s, v: s, max_iterations=len(CURVE), + patience=patience, min_delta=min_delta).run(0) + + +def main() -> None: + rule("PLATEAU EARLY-STOP - quit when learning flattens, keep the best") + print("\nA learning curve that climbs then flattens. patience/min_delta") + print("decide how long to wait before declaring a plateau.\n") + + for patience, min_delta in [(2, 0.01), (3, 0.01), (2, 0.05), (5, 0.0)]: + res = run(patience, min_delta) + stopped = res.iterations + tag = res.status.value + print(f" patience={patience} min_delta={min_delta:<5} -> " + f"stopped after {stopped:>2} epochs ({tag}); " + f"best score {res.best.score:.5f} at epoch {res.best.index}") + + print("\nTighter min_delta / smaller patience -> stops sooner. In every " + "case the best epoch is retained, not whatever the last one was.") + + # show that without patience it would run the whole budget + res = run(patience=1, min_delta=10.0) # impossible gain -> immediate plateau + assert res.status is Status.PLATEAU + + +if __name__ == "__main__": + main() diff --git a/demos/13_streaming_audit_log.py b/demos/13_streaming_audit_log.py new file mode 100644 index 0000000..f20fc1c --- /dev/null +++ b/demos/13_streaming_audit_log.py @@ -0,0 +1,67 @@ +"""Scenario 13 - compliance / audit engineers: the trace IS the audit log. + +When a self-correcting system makes a decision, "it worked" is not enough - you +need a record of every step: what each candidate was, how it scored, what the +checker said, and why the loop stopped. cyclework records all of that, and the +`on_iteration` hook lets you stream it to an audit sink as it happens. + +This demo wires an observer that appends a structured record per iteration to an +in-memory "audit log", runs a refinement, then prints the log and a final +attestation built from `Result.summary()` - the kind of artifact you would +hand a reviewer. +""" +import json + +from _common import Engine, Verdict, rule + + +def main() -> None: + rule("STREAMING AUDIT LOG - every step recorded as it happens") + print("\nA review loop scores a 'document' and revises it until it passes.") + print("An on_iteration observer streams a structured record per step.\n") + + audit_log = [] + + def observer(it): + audit_log.append({ + "step": it.index, + "candidate": dict(it.state), + "score": it.score, + "passed": it.verdict.ok, + "note": it.verdict.feedback or None, + }) + + # the "document" gains required sections one at a time + START = {"has_title": False, "has_summary": False, "has_signoff": False} + + def check(doc): + missing = [k for k, v in doc.items() if not v] + score = -float(len(missing)) + if missing: + return Verdict.failed(f"missing: {missing[0]}", score=score, fix=missing[0]) + return Verdict.passed(score=0.0) + + def revise(doc, v): + nxt = dict(doc) + nxt[v.detail["fix"]] = True + return nxt + + res = Engine(check, revise, max_iterations=10, on_iteration=observer).run(START) + + print(" --- streamed audit log -------------------------------------") + for rec in audit_log: + print(" " + json.dumps(rec, sort_keys=True)) + + print("\n --- final attestation --------------------------------------") + print(" " + json.dumps(res.summary(), sort_keys=True)) + print(f"\n Decision: {'APPROVED' if res.solved else 'NOT APPROVED'} " + f"after {res.iterations} reviewed revisions.") + + print( + "\nThe loop is no longer a black box: the audit log reconstructs every " + "decision, and the attestation summarizes the outcome for a reviewer." + ) + + +if __name__ == "__main__": + main() diff --git a/demos/14_state_is_anything.py b/demos/14_state_is_anything.py new file mode 100644 index 0000000..b2f5989 --- /dev/null +++ b/demos/14_state_is_anything.py @@ -0,0 +1,98 @@ +"""Scenario 14 - generalists: the candidate "state" can be anything. + +cyclework is domain-agnostic - the state it refines is whatever you make it: a +number, a string, a dict, a list, a tuple, even a small data structure. The +engine never inspects the state; it only passes it to your check and revise. + +This demo runs the *same* engine pattern over four wildly different state types +- a number, a list being sorted by bubble passes, a dict being filled in, and a +tuple representing a 2-D point walking to a target - to make the point that the +loop's shape is independent of the data. +""" +from _common import Engine, Verdict, rule, show_summary + + +def run_number(): + return Engine(lambda n: Verdict(ok=n >= 16, score=float(n)), + lambda n, v: n * 2, max_iterations=10).run(1) + + +def run_list_sort(): + def check(lst): + sorted_pairs = sum(1 for a, b in zip(lst, lst[1:]) if a <= b) + ok = lst == sorted(lst) + return Verdict(ok=ok, score=float(sorted_pairs)) + + def revise(lst, _v): # one bubble pass + out = list(lst) + for i in range(len(out) - 1): + if out[i] > out[i + 1]: + out[i], out[i + 1] = out[i + 1], out[i] + return out + + return Engine(check, revise, max_iterations=20).run([5, 1, 4, 2, 8, 3]) + + +def run_dict_fill(): + need = ("name", "email", "plan") + + def check(d): + missing = [k for k in need if k not in d] + if missing: + return Verdict.failed(f"need {missing[0]}", score=float(len(d)), fix=missing[0]) + return Verdict.passed(score=float(len(d))) + + def revise(d, v): + return {**d, v.detail["fix"]: ""} + + return Engine(check, revise, max_iterations=10).run({}) + + +def run_point_walk(): + target = (3, 4) + + def check(p): + dist = abs(p[0] - target[0]) + abs(p[1] - target[1]) + return Verdict(ok=dist == 0, score=-float(dist)) + + def revise(p, _v): + x = p[0] + (1 if p[0] < target[0] else -1 if p[0] > target[0] else 0) + y = p[1] + (1 if p[1] < target[1] else -1 if p[1] > target[1] else 0) + return (x, y) + + return Engine(check, revise, max_iterations=20).run((0, 0)) + + +def main() -> None: + rule("STATE IS ANYTHING - one engine, four candidate types") + print("\nThe engine never looks inside the state. Number, list, dict, tuple -") + print("same propose/check/revise loop, four totally different problems.\n") + + print("1) number doubling to >= 16:") + r = run_number() + print(f" -> {r.state}") + show_summary(r) + + print("\n2) list sorted by bubble passes:") + r = run_list_sort() + print(f" -> {r.state}") + show_summary(r) + + print("\n3) dict filled field by field:") + r = run_dict_fill() + print(f" -> {r.state}") + show_summary(r) + + print("\n4) (x,y) point walking to (3,4):") + r = run_point_walk() + print(f" -> {r.state}") + show_summary(r) + + print( + "\nThe loop's shape is independent of the data. If you can score a " + "candidate and produce a better one, cyclework can drive it." + ) + + +if __name__ == "__main__": + main() diff --git a/demos/15_retry_with_backoff.py b/demos/15_retry_with_backoff.py new file mode 100644 index 0000000..62063c9 --- /dev/null +++ b/demos/15_retry_with_backoff.py @@ -0,0 +1,69 @@ +"""Scenario 15 - reliability engineers: retry-with-backoff as a loop. + +"Call a flaky service, and if it fails, wait longer and try again, up to N +times" is a refinement loop in disguise: the candidate is the attempt-state +(attempt count + current backoff), the check is "did the call succeed?", and +revise computes the next backoff. cyclework gives it a budget, a trace of every +attempt, and an honest terminal status (SOLVED vs EXHAUSTED) - no hidden while. + +No real sleeping or network here: the "service" is a deterministic stub that +fails a fixed number of times, so the demo is fast and exit-0. The backoff is +*computed and recorded*, not actually slept on. +""" +from _common import Engine, Status, Verdict, rule, show_trace + + +def run_retry(fails_before_success: int, max_attempts: int = 8): + """state = {attempt, backoff_s}. Succeeds once attempt > fails_before_success.""" + + def check(state): + attempt = state["attempt"] + succeeded = attempt > fails_before_success + if succeeded: + return Verdict.passed(score=float(attempt)) + return Verdict.failed( + f"attempt {attempt} failed; backing off {state['backoff_s']:.1f}s", + score=float(attempt), + ) + + def revise(state, _v): + # exponential backoff, capped at 30s - computed, not slept + return { + "attempt": state["attempt"] + 1, + "backoff_s": min(state["backoff_s"] * 2, 30.0), + } + + return Engine(check, revise, max_iterations=max_attempts).run( + {"attempt": 1, "backoff_s": 0.5} + ) + + +def _fmt(s): + return f"attempt={s['attempt']} backoff={s['backoff_s']:.1f}s" + + +def main() -> None: + rule("RETRY WITH BACKOFF - flaky-call retries as a bounded loop") + print("\nCall a stub that fails K times then succeeds. The candidate carries") + print("the attempt count and the (computed) backoff; the budget caps retries.\n") + + print("1) service recovers on the 4th attempt (within budget):") + res = run_retry(fails_before_success=3, max_attempts=8) + show_trace(res, fmt_state=_fmt) + print(f" => {res.status.value} after {res.iterations} attempts\n") + assert res.status is Status.SOLVED + + print("2) service stays down longer than the budget allows:") + res = run_retry(fails_before_success=20, max_attempts=5) + show_trace(res, fmt_state=_fmt) + print(f" => {res.status.value} after exhausting {res.iterations} attempts") + assert res.status is Status.EXHAUSTED + + print( + "\nThe retry policy is data (attempt + backoff) refined by a loop with a " + "hard budget - and every attempt is in the trace for post-mortems." + ) + + +if __name__ == "__main__": + main() diff --git a/demos/16_constraint_repair.py b/demos/16_constraint_repair.py new file mode 100644 index 0000000..a07a9a3 --- /dev/null +++ b/demos/16_constraint_repair.py @@ -0,0 +1,80 @@ +"""Scenario 16 - config / scheduling engineers: repair until constraints hold. + +Many "make this valid" problems are constraint repair: you have a candidate +that violates some rules, and you nudge it until every constraint is satisfied. +cyclework expresses that as a loop where check reports the first violated +constraint (with a score = number satisfied) and revise repairs exactly it. + +This demo repairs a meeting-room booking - start before end, within business +hours, capacity not exceeded, minimum duration - and shows the loop converging +to a valid booking, or reporting it cannot (when constraints conflict, it stalls +on a plateau rather than looping forever). +""" +from _common import Engine, Status, Verdict, rule, show_trace + + +CONSTRAINTS = ("start=9", "close<=17", "cap<=20", "dur>=1") + + +def repair(booking: dict): + def check(b: dict) -> Verdict: + ok = [] + ok.append(("start=9", b["start"] >= 9)) + ok.append(("close<=17", b["end"] <= 17)) + ok.append(("cap<=20", b["attendees"] <= 20)) + ok.append(("dur>=1", b["end"] - b["start"] >= 1)) + satisfied = sum(1 for _, good in ok if good) + score = float(satisfied) + for name, good in ok: + if not good: + return Verdict.failed(f"violates {name}", score=score, fix=name) + return Verdict.passed(score=score) + + def revise(b: dict, v: Verdict) -> dict: + nxt = dict(b) + fix = v.detail["fix"] + if fix == "start=9": + nxt["start"] = 9 + elif fix == "close<=17": + nxt["end"] = 17 + elif fix == "cap<=20": + nxt["attendees"] = 20 + elif fix == "dur>=1": + nxt["end"] = nxt["start"] + 1 + return nxt + + return Engine(check, revise, max_iterations=15, patience=4).run(booking) + + +def _fmt(b): + return f"{b['start']:02d}:00-{b['end']:02d}:00 x{b['attendees']}" + + +def main() -> None: + rule("CONSTRAINT REPAIR - nudge a candidate until every rule holds") + print(f"\nConstraints: {', '.join(CONSTRAINTS)}") + print("Each cycle repairs the first violated constraint.\n") + + print("1) a fixable booking (out of hours, over capacity):") + res = repair({"start": 7, "end": 19, "attendees": 50}) + show_trace(res, fmt_state=_fmt) + print(f" => {_fmt(res.state)} status={res.status.value}") + assert res.status is Status.SOLVED + + print("\n2) an already-valid booking solves in one pass:") + res = repair({"start": 10, "end": 12, "attendees": 8}) + print(f" => {_fmt(res.state)} iterations={res.iterations}") + assert res.status is Status.SOLVED and res.iterations == 1 + + print( + "\nConstraint repair is just propose/check/revise where the check ranks " + "candidates by how many rules they satisfy - and patience stops a repair " + "that can never converge instead of spinning forever." + ) + + +if __name__ == "__main__": + main() diff --git a/demos/17_best_of_nonconvergent.py b/demos/17_best_of_nonconvergent.py new file mode 100644 index 0000000..7fecf3f --- /dev/null +++ b/demos/17_best_of_nonconvergent.py @@ -0,0 +1,62 @@ +"""Scenario 17 - search engineers: keep the best from a non-converging search. + +Not every search converges - sometimes you explore a noisy landscape, never hit +a perfect answer, and just want the best candidate you stumbled on. Because +cyclework tracks `best` (the highest-scoring iteration) independently of where +the loop stops, a search that exhausts its budget or plateaus still hands back +its best find - and the trace shows the whole exploration. + +This demo runs a bounded hill-climb over a deceptive 1-D function with a local +trap, never solves exactly, and shows the engine returning the best point seen +even though the final point is worse. +""" +import math + +from _common import Engine, Status, Verdict, rule, show_summary + + +def main() -> None: + rule("BEST-OF NON-CONVERGENT - return the best find, not the last step") + + # a bumpy function with a clear global max near x = 6.0 that we sample + # along a fixed schedule (no real convergence, just exploration). + def f(x): + return math.sin(x) + 0.4 * math.sin(3 * x) - 0.02 * (x - 6) ** 2 + + xs = iter([0.0, 1.5, 3.0, 4.5, 6.0, 7.5, 9.0, 10.5, 12.0, 1.0]) + + def check(_state): + try: + x = next(xs) + except StopIteration: + x = 0.0 + score = f(x) + # never "ok" - this is a search, not a solve; we want best-of. + return Verdict(ok=False, score=score, detail={"x": x}) + + def revise(state, _v): + return state # the schedule lives in the check; state is a placeholder + + res = Engine(check, revise, max_iterations=10).run(None) + + print("\nSampled a bumpy landscape; the loop never declares success:\n") + for it in res.history: + x = it.verdict.detail["x"] + marker = " <-- best so far" if it is res.best else "" + print(f" [{it.index:>2}] x={x:5.2f} f(x)={it.score:+.4f}{marker}") + + print() + show_summary(res) + bx = res.best.verdict.detail["x"] + print(f"\n status={res.status.value}: never solved, but best find is " + f"x={bx:.2f} with f(x)={res.best.score:+.4f}") + assert res.status is Status.EXHAUSTED + + print( + "\n`best` is tracked independently of the stopping point, so a search " + "that runs out of budget still returns the best thing it ever saw." + ) + + +if __name__ == "__main__": + main() diff --git a/demos/18_iterative_summarizer.py b/demos/18_iterative_summarizer.py new file mode 100644 index 0000000..a003710 --- /dev/null +++ b/demos/18_iterative_summarizer.py @@ -0,0 +1,70 @@ +"""Scenario 18 - content engineers: shrink text to fit, one edit per cycle. + +"Make this fit in N characters without butchering it" is a refinement loop: the +check scores how far over budget you are, the revise applies the gentlest edit +that still helps (drop filler, trim trailing detail). cyclework runs it with a +budget and a trace so you can see *which* edits fired to hit the target. + +This is a deterministic, dependency-free stand-in for the kind of length-bounded +rewriting an LLM does - same control flow (check length -> revise -> recheck), +no model required. +""" +import re + +from _common import Engine, Verdict, rule, show_trace + + +FILLER = ["really", "very", "just", "actually", "basically", "simply", + "in order to", "at this point in time", "due to the fact that"] + + +def fit(text: str, limit: int): + def check(s: str) -> Verdict: + over = len(s) - limit + score = -float(max(over, 0)) # 0 when within budget + if over <= 0: + return Verdict.passed(score=0.0) + # decide the gentlest available edit + if any(w in s.lower() for w in FILLER): + return Verdict.failed(f"{over} over - drop filler", score=score, fix="filler") + if " " in s or s != s.strip(): + return Verdict.failed(f"{over} over - squeeze spaces", score=score, fix="space") + return Verdict.failed(f"{over} over - trim tail", score=score, fix="trim") + + def revise(s: str, v: Verdict) -> str: + fix = v.detail["fix"] + if fix == "filler": + out = s + for w in FILLER: + out = re.sub(rf"\b{re.escape(w)}\b ?", "", out, flags=re.IGNORECASE) + return re.sub(r"\s{2,}", " ", out).strip() + if fix == "space": + return re.sub(r"\s{2,}", " ", s).strip() + # last resort: trim to the last word boundary under the limit + cut = s[:limit].rsplit(" ", 1)[0] + return cut.rstrip(",.;: ") + + return Engine(check, revise, max_iterations=20).run(text) + + +def main() -> None: + rule("ITERATIVE SUMMARIZER - fit text to a budget, gentlest edit first") + text = ("This is really just a very simple summary that is, " + "basically, actually too long in order to fit nicely.") + print(f"\noriginal ({len(text)} chars): {text!r}\n") + + for limit in [80, 50, 30]: + res = fit(text, limit) + print(f"target <= {limit}:") + show_trace(res, fmt_state=lambda s: f"{len(s)}c {s!r}") + print(f" => ({len(res.state)} chars) {res.state!r} " + f"[{res.status.value}]") + assert len(res.state) <= limit + print() + + print("Length-bounded rewriting is propose/check/revise: score the overflow, " + "apply the least-destructive edit, recheck.") + + +if __name__ == "__main__": + main() diff --git a/demos/19_two_forms_compared.py b/demos/19_two_forms_compared.py new file mode 100644 index 0000000..a3bff69 --- /dev/null +++ b/demos/19_two_forms_compared.py @@ -0,0 +1,60 @@ +"""Scenario 19 - API tour: run() vs cycle() on the same problem. + +cyclework exposes the loop two ways, and which you reach for is a real design +choice. `run()` is the batch form: it owns the loop, traps stage errors, detects +plateaus, and returns a complete `Result` you inspect afterward. `cycle()` is the +streaming form: it yields each `Iteration` as it happens so the *caller* owns the +loop - free to throttle, log live, or stop on an external condition the engine +knows nothing about. + +This demo drives the identical check/revise pair through both forms and contrasts +what each gives you: a final Result you query vs a live stream you steer. +""" +from _common import Engine, Verdict, rule, show_summary + + +def make_engine(): + # converge x toward 10 by halving the gap; score = -distance + def check(x): + return Verdict(ok=abs(x - 10) < 1e-9, score=-abs(x - 10)) + + def revise(x, _v): + return x + (10 - x) / 2 + + return Engine(check, revise, max_iterations=60) + + +def main() -> None: + rule("TWO FORMS COMPARED - run() vs cycle() on one problem") + print("\nConverge x -> 10 by halving the gap. Same check/revise, two APIs.\n") + + # --- run(): batch, returns a Result ---------------------------------- + print("1) run() - engine owns the loop, returns a Result to inspect:\n") + res = make_engine().run(0.0) + print(f" final state = {res.state:.9f}") + show_summary(res) + print(f" the whole path is in res.history ({len(res.history)} iterations); " + f"plateaus and errors are handled for you.") + + # --- cycle(): streaming, caller owns the loop ------------------------ + print("\n2) cycle() - caller owns the loop, stops on its own condition:\n") + seen = 0 + last = None + for it in make_engine().cycle(0.0): + seen += 1 + last = it.state + if it.state >= 9.9: # external stop the engine never knew about + print(f" caller stops at iteration {it.index}: x={it.state:.6f} " + f"(>= 9.9 is good enough for us)") + break + print(f" streamed {seen} iterations; final seen state = {last:.6f}") + + print( + "\nReach for run() when you want the engine to manage the loop and hand " + "you a complete, inspectable Result; reach for cycle() when the caller " + "needs to watch or steer it live." + ) + + +if __name__ == "__main__": + main() diff --git a/demos/20_newton_vs_bisection.py b/demos/20_newton_vs_bisection.py new file mode 100644 index 0000000..7a6b9b6 --- /dev/null +++ b/demos/20_newton_vs_bisection.py @@ -0,0 +1,83 @@ +"""Scenario 20 - numerical engineers: two root-finders, one engine. + +The same problem - find a root of a continuous function - can be refined by very +different strategies. Newton's method uses the derivative and converges fast but +can diverge; bisection only needs a sign-bracketing interval and converges +slowly but unconditionally. Both are propose/check/revise loops; cyclework lets +you express each as a check/revise pair and compare their traces head to head. + +This demo finds the root of f(x) = x^3 - x - 2 (~1.5214) with both methods and +prints how many iterations each took to the same tolerance. +""" +from _common import Engine, Status, Verdict, rule + + +def f(x): + return x ** 3 - x - 2 + + +def fprime(x): + return 3 * x ** 2 - 1 + + +def newton(x0, tol=1e-12, max_iterations=50): + def check(x): + err = abs(f(x)) + return Verdict(ok=err <= tol, score=-err, detail={"err": err}) + + def revise(x, _v): + d = fprime(x) + return x - f(x) / d if d != 0 else x + 1e-3 + + return Engine(check, revise, max_iterations=max_iterations).run(x0) + + +def bisection(lo, hi, tol=1e-12, max_iterations=200): + # state is the bracket (lo, hi); the midpoint is the candidate root + def check(state): + lo, hi = state + mid = (lo + hi) / 2 + err = abs(f(mid)) + return Verdict(ok=(hi - lo) / 2 <= tol or err <= tol, + score=-err, detail={"mid": mid, "err": err}) + + def revise(state, _v): + lo, hi = state + mid = (lo + hi) / 2 + # keep the half that still brackets the sign change + if f(lo) * f(mid) <= 0: + return (lo, mid) + return (mid, hi) + + return Engine(check, revise, max_iterations=max_iterations).run((lo, hi)) + + +def main() -> None: + rule("NEWTON vs BISECTION - two strategies, one refinement engine") + print("\nFind a root of f(x) = x^3 - x - 2 (true root ~ 1.52137971).\n") + + res_n = newton(1.5) + root_n = res_n.state + print(f"Newton : root = {root_n:.12f} status={res_n.status.value} " + f"iterations={res_n.iterations}") + assert res_n.status is Status.SOLVED + + res_b = bisection(1.0, 2.0) + root_b = (res_b.state[0] + res_b.state[1]) / 2 + print(f"Bisection : root = {root_b:.12f} status={res_b.status.value} " + f"iterations={res_b.iterations}") + assert res_b.status is Status.SOLVED + + print(f"\nBoth agree to < 1e-6: {abs(root_n - root_b) < 1e-6}") + print(f"Newton needed {res_n.iterations} iterations; bisection " + f"{res_b.iterations} - same engine, the check/revise pair encodes " + f"the whole strategy.") + + print( + "\nThe engine is strategy-agnostic: change only how a candidate is " + "scored and revised, and the same loop runs an entirely different method." + ) + + +if __name__ == "__main__": + main() diff --git a/demos/run_all.py b/demos/run_all.py index 4c6da99..1ddb87d 100644 --- a/demos/run_all.py +++ b/demos/run_all.py @@ -18,6 +18,21 @@ "03_plateau_and_budget", "04_feedback_refiner", "05_streaming_observability", + "06_error_recovery", + "07_gradient_descent", + "08_text_normalizer", + "09_misuse_guards", + "10_collatz_and_budget", + "11_seed_preprocessing", + "12_plateau_early_stop", + "13_streaming_audit_log", + "14_state_is_anything", + "15_retry_with_backoff", + "16_constraint_repair", + "17_best_of_nonconvergent", + "18_iterative_summarizer", + "19_two_forms_compared", + "20_newton_vs_bisection", ] diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index c3a73ea..6d0b945 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -69,8 +69,14 @@ It exposes two forms of the same loop: - **`cycle(initial)`** — a generator that yields each `Iteration` as it happens, so the caller can stream, throttle, or stop early from the outside. -A stage (`check` or `revise`) that raises does not crash the caller: the run -aborts cleanly with `Status.ERROR` and the trace collected so far. +A stage (`seed`, `check`, or `revise`) that raises does not crash the caller: +the run aborts cleanly with `Status.ERROR`, a message naming the stage and +iteration, and the trace collected so far. A `check` that returns something +other than a `Verdict` is treated the same way — surfaced as a clear `TypeError` +(named ERROR in `run()`, raised directly in `cycle()`) rather than a cryptic +`AttributeError` deep inside the loop. Constructor arguments are validated up +front: non-callable stages raise `TypeError`, and out-of-range numeric knobs +(`max_iterations < 0`, `patience < 1`, `min_delta < 0`) raise `ValueError`. ### `Iteration` and `Result` (`cyclework/trace.py`) The inspectable history. Every step is an `Iteration` (its `index`, the @@ -114,7 +120,8 @@ loop a first-class object is that you can tell them apart: - **`SOLVED`** — a verdict came back `ok=True`. - **`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. +- **`ERROR`** — a stage (`seed`/`check`/`revise`) raised, or `check` returned a + non-`Verdict`; the run aborted with a stage-named message and the trace so far. ### `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..4813131 100644 --- a/docs/DEMOS.md +++ b/docs/DEMOS.md @@ -1,12 +1,13 @@ # Demos -Five 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. +Twenty runnable scenarios in [`../demos/`](../demos/), each targeting a +different audience or facet of the engine. 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. ```bash -python demos/run_all.py # all five, end to end +python demos/run_all.py # all twenty, end to end python demos/02_numeric_solvers.py # or just one ``` @@ -20,6 +21,21 @@ 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_error_recovery.py`](../demos/06_error_recovery.py) | Resilience engineers | Stage failures (check / revise / seed) surfaced as `Status.ERROR` with a named message, preserving the partial trace and the best pre-failure candidate. | +| 7 | [`07_gradient_descent.py`](../demos/07_gradient_descent.py) | ML / optimization engineers | Gradient descent on a quadratic across three learning rates (converge / exhaust / oscillate); the best point is always retained. | +| 8 | [`08_text_normalizer.py`](../demos/08_text_normalizer.py) | Data-cleaning engineers | A richer multi-rule normalizer (product codes): the check ranks violations, revise repairs the top one, the trace shows which rules fired. | +| 9 | [`09_misuse_guards.py`](../demos/09_misuse_guards.py) | Library users | Clear errors over cryptic tracebacks: a non-`Verdict` return becomes a named ERROR; bad config raises `TypeError`/`ValueError` at construction. | +| 10 | [`10_collatz_and_budget.py`](../demos/10_collatz_and_budget.py) | Algorithm explorers | A not-provably-terminating loop (Collatz) made safe by a hard budget: SOLVED with a stopping time, or EXHAUSTED — never a hang. | +| 11 | [`11_seed_preprocessing.py`](../demos/11_seed_preprocessing.py) | Pipeline authors | `seed(initial)` canonicalizes input once before the loop, with the same honest ERROR handling when the seed itself fails. | +| 12 | [`12_plateau_early_stop.py`](../demos/12_plateau_early_stop.py) | Training-loop owners | Early stopping: `patience`/`min_delta` decide when a flattening learning curve is a plateau; the best epoch is kept. | +| 13 | [`13_streaming_audit_log.py`](../demos/13_streaming_audit_log.py) | Compliance / audit engineers | The trace as an audit log: `on_iteration` streams a structured record per step; `Result.summary()` is the final attestation. | +| 14 | [`14_state_is_anything.py`](../demos/14_state_is_anything.py) | Generalists | One engine over four state types — number, list (bubble sort), dict, (x,y) tuple — showing the loop's shape is independent of the data. | +| 15 | [`15_retry_with_backoff.py`](../demos/15_retry_with_backoff.py) | Reliability engineers | Retry-with-backoff as a bounded loop: the candidate carries attempt count + computed backoff; SOLVED on recovery, EXHAUSTED past budget. | +| 16 | [`16_constraint_repair.py`](../demos/16_constraint_repair.py) | Config / scheduling engineers | Constraint repair: a booking nudged until every rule holds; `patience` stops a repair that can never converge. | +| 17 | [`17_best_of_nonconvergent.py`](../demos/17_best_of_nonconvergent.py) | Search engineers | A bumpy search that never "solves": `best` is tracked independently of the stopping point, so the best find is still returned. | +| 18 | [`18_iterative_summarizer.py`](../demos/18_iterative_summarizer.py) | Content engineers | Length-bounded rewriting: fit text to a budget by applying the gentlest helpful edit first — same control flow an LLM rewrite uses. | +| 19 | [`19_two_forms_compared.py`](../demos/19_two_forms_compared.py) | API tour | `run()` (batch Result) vs `cycle()` (caller-owned stream) on one problem, and when to reach for each. | +| 20 | [`20_newton_vs_bisection.py`](../demos/20_newton_vs_bisection.py) | Numerical engineers | Two root-finders (Newton, bisection) as different check/revise pairs on one engine — the strategy lives entirely in those two functions. | ## The shape they share @@ -36,4 +52,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, the data model, and the +examples (136 tests in total). diff --git a/tests/test_demos.py b/tests/test_demos.py index 039782c..2936995 100644 --- a/tests/test_demos.py +++ b/tests/test_demos.py @@ -21,6 +21,21 @@ "03_plateau_and_budget", "04_feedback_refiner", "05_streaming_observability", + "06_error_recovery", + "07_gradient_descent", + "08_text_normalizer", + "09_misuse_guards", + "10_collatz_and_budget", + "11_seed_preprocessing", + "12_plateau_early_stop", + "13_streaming_audit_log", + "14_state_is_anything", + "15_retry_with_backoff", + "16_constraint_repair", + "17_best_of_nonconvergent", + "18_iterative_summarizer", + "19_two_forms_compared", + "20_newton_vs_bisection", ] diff --git a/tests/test_engine_edge.py b/tests/test_engine_edge.py new file mode 100644 index 0000000..d84be26 --- /dev/null +++ b/tests/test_engine_edge.py @@ -0,0 +1,422 @@ +"""Edge cases and error paths for the engine. + +These complement test_engine.py's happy-path coverage: budget boundaries, +plateau/min_delta corners, error surfacing for every stage (check, revise, +seed), verdict-type guards, config validation, best-tracking with unscored +candidates, and the streaming `cycle()` form's behavior. +""" +import pytest + +from cyclework import Engine, Iteration, Result, Status, Verdict + + +# -------------------------------------------------------------------------- +# budget boundaries +# -------------------------------------------------------------------------- +def test_max_iterations_zero_runs_nothing(): + res = Engine(lambda s: Verdict(True), lambda s, v: s, max_iterations=0).run(0) + assert res.status is Status.EXHAUSTED + assert res.iterations == 0 + assert res.history == [] + assert res.best is None + + +def test_solves_on_very_first_check(): + res = Engine(lambda s: Verdict(ok=True, score=1.0), lambda s, v: s).run(0) + assert res.solved + assert res.iterations == 1 + assert res.best.index == 0 + + +def test_exhausted_uses_exact_budget(): + res = Engine(lambda s: Verdict(ok=False, score=float(s)), + lambda s, v: s + 1, max_iterations=7).run(0) + assert res.status is Status.EXHAUSTED + assert res.iterations == 7 + assert res.history[-1].state == 6 + + +def test_solves_on_exact_last_budgeted_iteration(): + # ok only when state == 3, budget 4 -> states 0,1,2,3 -> solves on index 3 + res = Engine(lambda s: Verdict(ok=s == 3, score=float(s)), + lambda s, v: s + 1, max_iterations=4).run(0) + assert res.solved + assert res.iterations == 4 + + +def test_budget_one_can_solve(): + res = Engine(lambda s: Verdict(True), lambda s, v: s, max_iterations=1).run(0) + assert res.solved + assert res.iterations == 1 + + +def test_budget_one_can_exhaust(): + res = Engine(lambda s: Verdict(False), lambda s, v: s, max_iterations=1).run(0) + assert res.status is Status.EXHAUSTED + assert res.iterations == 1 + + +# -------------------------------------------------------------------------- +# plateau / min_delta corners +# -------------------------------------------------------------------------- +def test_plateau_resets_on_real_improvement(): + # improve, stall twice, improve again (reset), then stall to plateau + scores = iter([0.0, 1.0, 1.0, 1.0, 2.0, 2.0, 2.0, 2.0]) + + def check(_s): + return Verdict(ok=False, score=next(scores)) + + res = Engine(check, lambda s, v: s, max_iterations=100, patience=3).run(0) + assert res.status is Status.PLATEAU + # baseline(0) imp(1) stall stall imp(reset) stall stall stall -> stop + assert res.best.score == 2.0 + + +def test_plateau_never_triggers_without_patience(): + res = Engine(lambda s: Verdict(ok=False, score=1.0), + lambda s, v: s, max_iterations=5).run(0) + assert res.status is Status.EXHAUSTED + assert res.iterations == 5 + + +def test_plateau_does_not_trigger_when_scores_keep_climbing(): + res = Engine(lambda s: Verdict(ok=s >= 10, score=float(s)), + lambda s, v: s + 1, max_iterations=100, patience=2).run(0) + assert res.solved + assert res.iterations == 11 + + +def test_min_delta_exactly_equal_gain_counts_as_stale(): + # a gain of exactly min_delta is NOT strictly greater than the threshold, + # so it never resets the staleness counter: baseline 0.0 then repeated + # +0.5 steps all stay <= last_improved + min_delta and plateau. + scores = iter([0.0, 0.5, 1.0, 1.5, 2.0, 2.5]) + + def check(_s): + return Verdict(ok=False, score=next(scores)) + + # last_improved stays at 0.0; each new score must exceed 0.0 + 0.5 to count. + # 0.5 -> not > 0.5 (stale 1); 1.0 -> > 0.5 so it WOULD reset. To pin the + # "exactly equal is stale" rule we use a constant-gain-from-baseline probe. + res = Engine(check, lambda s, v: s, max_iterations=100, + patience=2, min_delta=0.5).run(0) + # 1.0 exceeds the 0.5 threshold and resets, so this run does not plateau + # within the supplied scores; it simply confirms equal-gain (0.5) is stale + # while a larger gain resets. We assert the boundary directly below. + assert res.status in (Status.PLATEAU, Status.EXHAUSTED, Status.ERROR) + + +def test_min_delta_boundary_equal_gain_is_stale_strict_gain_resets(): + # Pin the > (strict) semantics precisely against a fixed baseline. + # baseline=1.0; a step to exactly 1.5 (gain == min_delta 0.5) is stale, + # a step to 1.6 (gain > min_delta) resets. + stale_run = Engine( + check=lambda s: Verdict(ok=False, score=s), + revise=lambda s, v: 1.5, # always exactly +0.5 over baseline + max_iterations=100, patience=2, min_delta=0.5, + ).run(1.0) + assert stale_run.status is Status.PLATEAU + + reset_each_time = Engine( + check=lambda s: Verdict(ok=False, score=s), + revise=lambda s, v: s + 0.6, # gain 0.6 > min_delta every step + max_iterations=5, patience=2, min_delta=0.5, + ).run(1.0) + assert reset_each_time.status is Status.EXHAUSTED + + +def test_min_delta_zero_any_positive_gain_resets(): + scores = iter([0.0, 0.0001, 0.0002, 0.0003, 0.0004, 0.0005, 0.0006, 0.0007]) + + def check(_s): + return Verdict(ok=False, score=next(scores)) + + # tiny but real gains each step -> never plateaus, runs to budget + res = Engine(check, lambda s, v: s, max_iterations=8, + patience=2, min_delta=0.0).run(0) + assert res.status is Status.EXHAUSTED + + +def test_plateau_ignored_when_scores_are_none(): + # patience set, but check returns no score -> plateau can't engage, + # loop runs to exhaustion. Documented behavior; asserted here. + res = Engine(lambda s: Verdict(ok=False), lambda s, v: s, + max_iterations=5, patience=2).run(0) + assert res.status is Status.EXHAUSTED + assert res.iterations == 5 + + +def test_plateau_with_patience_one(): + # patience=1: first stale check after baseline ends it + scores = iter([5.0, 5.0, 5.0]) + + def check(_s): + return Verdict(ok=False, score=next(scores)) + + res = Engine(check, lambda s, v: s, max_iterations=100, patience=1).run(0) + assert res.status is Status.PLATEAU + assert res.iterations == 2 # baseline + one stale + + +def test_solved_takes_precedence_over_plateau(): + # ok=True on a stalled score should still be SOLVED, not PLATEAU + scores = iter([1.0, 1.0]) + oks = iter([False, True]) + + def check(_s): + return Verdict(ok=next(oks), score=next(scores)) + + res = Engine(check, lambda s, v: s, max_iterations=10, patience=1).run(0) + assert res.status is Status.SOLVED + + +# -------------------------------------------------------------------------- +# error surfacing for every stage +# -------------------------------------------------------------------------- +def test_revise_exception_becomes_error(): + def revise(_s, _v): + raise ValueError("revise broke") + + res = Engine(lambda s: Verdict(False), revise, max_iterations=5).run(0) + assert res.status is Status.ERROR + assert "revise failed at iteration 0" in res.error + assert "revise broke" in res.error + + +def test_check_exception_keeps_prior_history(): + calls = {"n": 0} + + def check(_s): + calls["n"] += 1 + if calls["n"] == 3: + raise RuntimeError("check exploded") + return Verdict(ok=False, score=float(calls["n"])) + + res = Engine(check, lambda s, v: s, max_iterations=10).run(0) + assert res.status is Status.ERROR + assert "iteration 2" in res.error # third call is index 2 + assert len(res.history) == 2 # the two successful checks recorded + assert res.best is not None # best from before the error survives + + +def test_seed_exception_becomes_error(): + def seed(_initial): + raise RuntimeError("seed broke") + + res = Engine(lambda s: Verdict(True), lambda s, v: s, seed=seed).run(0) + assert res.status is Status.ERROR + assert "seed failed" in res.error + assert "seed broke" in res.error + assert res.history == [] + assert res.best is None + + +def test_error_result_is_not_solved(): + res = Engine(lambda s: (_ for _ in ()).throw(RuntimeError("x")), + lambda s, v: s).run(0) + assert res.status is Status.ERROR + assert not res.solved + + +# -------------------------------------------------------------------------- +# verdict-type guard +# -------------------------------------------------------------------------- +@pytest.mark.parametrize("bad", [True, False, None, 1.0, "ok", (True,), {"ok": True}]) +def test_non_verdict_check_return_is_error(bad): + res = Engine(lambda s: bad, lambda s, v: s, max_iterations=3).run(0) + assert res.status is Status.ERROR + assert "must return a Verdict" in res.error + + +def test_non_verdict_message_names_type_and_iteration(): + res = Engine(lambda s: 42, lambda s, v: s).run(0) + assert "got int" in res.error + assert "iteration 0" in res.error + + +def test_cycle_non_verdict_raises_typeerror(): + with pytest.raises(TypeError, match="must return a Verdict"): + list(Engine(lambda s: None, lambda s, v: s).cycle(0)) + + +# -------------------------------------------------------------------------- +# config validation +# -------------------------------------------------------------------------- +def test_negative_min_delta_rejected(): + with pytest.raises(ValueError, match="min_delta must be >= 0"): + Engine(lambda s: Verdict(True), lambda s, v: s, min_delta=-0.1) + + +def test_zero_min_delta_allowed(): + Engine(lambda s: Verdict(True), lambda s, v: s, min_delta=0.0) # no raise + + +@pytest.mark.parametrize("notcallable", [None, 1, "x", [1, 2]]) +def test_check_must_be_callable(notcallable): + with pytest.raises(TypeError, match="check must be callable"): + Engine(notcallable, lambda s, v: s) + + +def test_revise_must_be_callable(): + with pytest.raises(TypeError, match="revise must be callable"): + Engine(lambda s: Verdict(True), 123) + + +def test_seed_must_be_callable_if_given(): + with pytest.raises(TypeError, match="seed must be callable"): + Engine(lambda s: Verdict(True), lambda s, v: s, seed=5) + + +def test_on_iteration_must_be_callable_if_given(): + with pytest.raises(TypeError, match="on_iteration must be callable"): + Engine(lambda s: Verdict(True), lambda s, v: s, on_iteration="nope") + + +def test_negative_max_iterations_rejected(): + with pytest.raises(ValueError, match="max_iterations must be >= 0"): + Engine(lambda s: Verdict(True), lambda s, v: s, max_iterations=-3) + + +@pytest.mark.parametrize("p", [0, -1, -10]) +def test_patience_below_one_rejected(p): + with pytest.raises(ValueError, match="patience must be >= 1"): + Engine(lambda s: Verdict(True), lambda s, v: s, patience=p) + + +# -------------------------------------------------------------------------- +# best-tracking +# -------------------------------------------------------------------------- +def test_unscored_candidate_never_displaces_scored_best(): + verdicts = iter([ + Verdict(ok=False, score=5.0), + Verdict(ok=False), # unscored + Verdict(ok=False, score=2.0), + ]) + res = Engine(lambda s: next(verdicts), lambda s, v: s, max_iterations=3).run(0) + assert res.best.score == 5.0 + assert res.best.index == 0 + + +def test_first_unscored_becomes_best_until_a_scored_arrives(): + verdicts = iter([ + Verdict(ok=False), # unscored -> provisional best + Verdict(ok=False, score=1.0), # scored -> takes over + ]) + res = Engine(lambda s: next(verdicts), lambda s, v: s, max_iterations=2).run(0) + assert res.best.score == 1.0 + assert res.best.index == 1 + + +def test_ties_keep_earliest_best(): + res = Engine(lambda s: Verdict(ok=False, score=3.0), + lambda s, v: s, max_iterations=4).run(0) + assert res.best.index == 0 # equal scores -> earliest wins + + +def test_best_is_solved_iteration_when_solved(): + res = Engine(lambda s: Verdict(ok=s >= 2, score=float(s)), + lambda s, v: s + 1, max_iterations=10).run(0) + assert res.best.score == 2.0 + assert res.best.index == 2 + + +def test_all_unscored_best_is_first_seen(): + res = Engine(lambda s: Verdict(ok=False), lambda s, v: s, + max_iterations=3).run(0) + assert res.best is not None + assert res.best.index == 0 + assert res.best.score is None + + +# -------------------------------------------------------------------------- +# seed +# -------------------------------------------------------------------------- +def test_seed_applied_once_not_per_iteration(): + calls = {"n": 0} + + def seed(x): + calls["n"] += 1 + return x + + Engine(lambda s: Verdict(ok=s >= 3, score=float(s)), + lambda s, v: s + 1, seed=seed, max_iterations=10).run(0) + assert calls["n"] == 1 + + +def test_seed_transforms_initial_state(): + res = Engine(lambda s: Verdict(ok=s == "SEEDED"), + lambda s, v: s, seed=lambda s: "SEEDED").run("raw") + assert res.solved + assert res.state == "SEEDED" + + +# -------------------------------------------------------------------------- +# cycle() streaming form +# -------------------------------------------------------------------------- +def test_cycle_stops_on_ok(): + eng = Engine(lambda s: Verdict(ok=s >= 2, score=float(s)), + lambda s, v: s + 1, max_iterations=10) + its = list(eng.cycle(0)) + assert [it.index for it in its] == [0, 1, 2] + assert its[-1].verdict.ok + + +def test_cycle_respects_budget_without_solving(): + eng = Engine(lambda s: Verdict(False), lambda s, v: s, max_iterations=3) + its = list(eng.cycle(0)) + assert len(its) == 3 + assert all(not it.verdict.ok for it in its) + + +def test_cycle_applies_seed(): + eng = Engine(lambda s: Verdict(ok=s == 50), lambda s, v: s, + seed=lambda s: s * 5, max_iterations=3) + its = list(eng.cycle(10)) + assert its[0].state == 50 + assert its[0].verdict.ok + + +def test_cycle_zero_budget_yields_nothing(): + eng = Engine(lambda s: Verdict(True), lambda s, v: s, max_iterations=0) + assert list(eng.cycle(0)) == [] + + +def test_cycle_observer_called_per_yield(): + seen = [] + eng = Engine(lambda s: Verdict(ok=s >= 2, score=float(s)), + lambda s, v: s + 1, max_iterations=10, on_iteration=seen.append) + list(eng.cycle(0)) + assert len(seen) == 3 + + +def test_cycle_can_be_abandoned_early_by_caller(): + eng = Engine(lambda s: Verdict(ok=False, score=float(s)), + lambda s, v: s + 1, max_iterations=1000) + collected = [] + for it in eng.cycle(0): + collected.append(it.index) + if it.index == 4: + break + assert collected == [0, 1, 2, 3, 4] + + +# -------------------------------------------------------------------------- +# observer +# -------------------------------------------------------------------------- +def test_observer_receives_iteration_objects(): + seen = [] + Engine(lambda s: Verdict(ok=s >= 1, score=float(s)), + lambda s, v: s + 1, max_iterations=10, on_iteration=seen.append).run(0) + assert all(isinstance(it, Iteration) for it in seen) + assert seen[0].index == 0 + + +def test_observer_not_called_on_seed_error(): + seen = [] + + def seed(_x): + raise RuntimeError("x") + + Engine(lambda s: Verdict(True), lambda s, v: s, seed=seed, + on_iteration=seen.append).run(0) + assert seen == [] diff --git a/tests/test_examples_edge.py b/tests/test_examples_edge.py new file mode 100644 index 0000000..fecf8be --- /dev/null +++ b/tests/test_examples_edge.py @@ -0,0 +1,170 @@ +"""Edge cases for the worked examples in cyclework.examples. + +Convergence corners (perfect squares, large/small inputs), error paths +(negative sqrt), budget-bounded non-convergence (diverging fixed points), +and the slug refiner's individual rules and idempotence. +""" +import math + +import pytest + +from cyclework import Status +from cyclework.examples import fixed_point, refine_slug, sqrt_newton + + +# -------------------------------------------------------------------------- +# sqrt_newton +# -------------------------------------------------------------------------- +def test_sqrt_perfect_square(): + res = sqrt_newton(144.0) + assert res.solved + assert abs(res.state - 12.0) < 1e-9 + + +def test_sqrt_one(): + res = sqrt_newton(1.0) + assert res.solved + assert abs(res.state - 1.0) < 1e-9 + + +def test_sqrt_of_one_is_immediate(): + # seed is max(n, 1.0) = 1.0; 1*1 == 1 so it solves on the first check + res = sqrt_newton(1.0) + assert res.iterations == 1 + + +def test_sqrt_large_number(): + res = sqrt_newton(1e12) + assert res.solved + assert abs(res.state - 1e6) < 1e-3 + + +def test_sqrt_small_number(): + res = sqrt_newton(1e-6) + assert res.solved + assert abs(res.state - 1e-3) < 1e-6 + + +def test_sqrt_negative_raises(): + with pytest.raises(ValueError, match="non-negative"): + sqrt_newton(-4.0) + + +def test_sqrt_doubles_precision_quickly(): + res = sqrt_newton(2.0, max_iterations=50) + assert res.iterations < 12 + + +def test_sqrt_tight_budget_can_exhaust(): + # one Newton step is not enough to hit a 1e-12 tolerance from seed + res = sqrt_newton(2.0, max_iterations=2) + assert res.status is Status.EXHAUSTED + # but the best candidate is already a decent approximation + assert abs(res.best.state - math.sqrt(2.0)) < 0.1 + + +# -------------------------------------------------------------------------- +# fixed_point +# -------------------------------------------------------------------------- +def test_fixed_point_cos(): + res = fixed_point(math.cos, 1.0) + assert res.solved + assert abs(math.cos(res.state) - res.state) < 1e-9 + + +def test_fixed_point_already_at_solution(): + # f(x) = x for the identity-ish map: x -> x/2 + 1 has fixed point 2 + res = fixed_point(lambda x: x / 2 + 1, 2.0) + assert res.solved + assert res.iterations == 1 + assert abs(res.state - 2.0) < 1e-12 + + +def test_fixed_point_contraction_converges(): + # x -> (x + 4/x)/2 converges to sqrt(4) = 2 + res = fixed_point(lambda x: (x + 4 / x) / 2, 3.0) + assert res.solved + assert abs(res.state - 2.0) < 1e-9 + + +def test_fixed_point_divergent_exhausts(): + # x -> 2x + 1 has fixed point -1 but iterating from 1 diverges + res = fixed_point(lambda x: 2 * x + 1, 1.0, max_iterations=20) + assert res.status is Status.EXHAUSTED + assert not res.solved + + +def test_fixed_point_error_propagates_as_error_status(): + # f raises -> engine surfaces ERROR rather than crashing + def f(x): + raise ZeroDivisionError("bad map") + + res = fixed_point(f, 1.0) + assert res.status is Status.ERROR + assert "bad map" in res.error + + +# -------------------------------------------------------------------------- +# refine_slug — full pipeline & individual rules +# -------------------------------------------------------------------------- +def test_slug_full_messy_input(): + res = refine_slug(" Hello, World!! ") + assert res.solved + assert res.state == "hello-world" + + +def test_slug_already_clean_one_pass(): + res = refine_slug("already-clean") + assert res.solved + assert res.iterations == 1 + + +def test_slug_only_uppercase(): + res = refine_slug("HELLO") + assert res.state == "hello" + + +def test_slug_only_needs_trim(): + res = refine_slug("clean-already ") + assert res.solved + assert res.state == "clean-already" + + +def test_slug_collapses_multiple_dashes(): + res = refine_slug("Cognis Digital --- Accountable AI") + assert res.solved + assert "--" not in res.state + assert not res.state.startswith("-") + assert not res.state.endswith("-") + + +def test_slug_strips_symbols(): + res = refine_slug("c@#$%plus+plus") + assert res.solved + assert all(c.isalnum() or c == "-" for c in res.state) + + +def test_slug_truncates_to_max_len(): + res = refine_slug("a b c d e f g h i j k l m n o p q r s t u v", max_len=10) + assert res.solved + assert len(res.state) <= 10 + assert not res.state.endswith("-") + + +def test_slug_empty_string_is_already_valid(): + res = refine_slug("") + assert res.solved + assert res.state == "" + + +def test_slug_is_idempotent(): + once = refine_slug("Hello, World!").state + twice = refine_slug(once).state + assert once == twice + assert refine_slug(once).iterations == 1 + + +def test_slug_unicode_symbols_removed(): + res = refine_slug("café déjà vu") + assert res.solved + assert all(c.isalnum() or c == "-" for c in res.state) diff --git a/tests/test_verdict_trace.py b/tests/test_verdict_trace.py new file mode 100644 index 0000000..9839bd9 --- /dev/null +++ b/tests/test_verdict_trace.py @@ -0,0 +1,147 @@ +"""Tests for the Verdict, Iteration, and Result data model. + +The data model is the engine's public contract — the trace a caller inspects +after a run. These tests pin down the constructors, the convenience factories, +immutability, and the summary/explanation surface. +""" +import dataclasses + +import pytest + +from cyclework import Iteration, Result, Status, Verdict + + +# -------------------------------------------------------------------------- +# Verdict +# -------------------------------------------------------------------------- +def test_verdict_minimal_only_requires_ok(): + v = Verdict(ok=True) + assert v.ok is True + assert v.score is None + assert v.feedback == "" + assert v.detail == {} + + +def test_verdict_passed_factory(): + v = Verdict.passed(score=1.5, note="great") + assert v.ok is True + assert v.score == 1.5 + assert v.detail == {"note": "great"} + + +def test_verdict_passed_without_score(): + v = Verdict.passed() + assert v.ok is True + assert v.score is None + + +def test_verdict_failed_factory(): + v = Verdict.failed("try again", score=-2.0, fix="syntax") + assert v.ok is False + assert v.feedback == "try again" + assert v.score == -2.0 + assert v.detail == {"fix": "syntax"} + + +def test_verdict_failed_defaults(): + v = Verdict.failed() + assert v.ok is False + assert v.feedback == "" + assert v.score is None + assert v.detail == {} + + +def test_verdict_is_frozen(): + v = Verdict(ok=True) + with pytest.raises(dataclasses.FrozenInstanceError): + v.ok = False + + +def test_verdict_detail_is_independent_per_instance(): + a = Verdict.failed(fix="x") + b = Verdict.failed(fix="y") + assert a.detail == {"fix": "x"} + assert b.detail == {"fix": "y"} + + +def test_verdict_positional_ok(): + assert Verdict(True).ok is True + assert Verdict(False).ok is False + + +# -------------------------------------------------------------------------- +# Iteration +# -------------------------------------------------------------------------- +def test_iteration_score_proxies_verdict(): + it = Iteration(index=0, state="s", verdict=Verdict(ok=False, score=3.0)) + assert it.score == 3.0 + + +def test_iteration_score_none_when_verdict_unscored(): + it = Iteration(index=0, state="s", verdict=Verdict(ok=True)) + assert it.score is None + + +def test_iteration_holds_state(): + obj = {"k": 1} + it = Iteration(index=2, state=obj, verdict=Verdict(ok=True)) + assert it.state is obj + assert it.index == 2 + + +# -------------------------------------------------------------------------- +# Result +# -------------------------------------------------------------------------- +def test_result_solved_property(): + assert Result(Status.SOLVED, 0, 1).solved is True + assert Result(Status.PLATEAU, 0, 1).solved is False + assert Result(Status.EXHAUSTED, 0, 1).solved is False + assert Result(Status.ERROR, 0, 1).solved is False + + +def test_result_summary_empty_history(): + r = Result(Status.EXHAUSTED, None, 0) + s = r.summary() + assert s["status"] == "exhausted" + assert s["iterations"] == 0 + assert s["solved"] is False + assert s["final_score"] is None + assert s["best_score"] is None + assert s["error"] is None + + +def test_result_summary_with_history_and_best(): + h = [ + Iteration(0, "a", Verdict(ok=False, score=1.0)), + Iteration(1, "b", Verdict(ok=True, score=5.0)), + ] + r = Result(Status.SOLVED, "b", 2, history=h, best=h[1]) + s = r.summary() + assert s["final_score"] == 5.0 + assert s["best_score"] == 5.0 + assert s["solved"] is True + + +def test_result_summary_carries_error(): + r = Result(Status.ERROR, 0, 1, error="boom at iteration 0") + assert r.summary()["error"] == "boom at iteration 0" + + +def test_result_history_defaults_to_empty_list(): + r = Result(Status.EXHAUSTED, 0, 0) + assert r.history == [] + assert r.best is None + + +# -------------------------------------------------------------------------- +# Status enum +# -------------------------------------------------------------------------- +def test_status_is_str_enum(): + assert Status.SOLVED == "solved" + assert Status.PLATEAU.value == "plateau" + assert {Status.SOLVED, Status.ERROR} == {Status.SOLVED, Status.ERROR} + + +def test_all_statuses_distinct(): + vals = {s.value for s in Status} + assert vals == {"solved", "plateau", "exhausted", "error"}