diff --git a/wizard-work/null/budgeted-planner/README-budgeted-planner.md b/wizard-work/null/budgeted-planner/README-budgeted-planner.md new file mode 100644 index 0000000..bcf6d8b --- /dev/null +++ b/wizard-work/null/budgeted-planner/README-budgeted-planner.md @@ -0,0 +1,66 @@ +# budgeted_planner — exact optimal task selection under a token budget + +Lane: Codeforge. Artifact: `budgeted_planner.py` (+ `test_budgeted_planner.py` test suite). +Pure stdlib, no dependencies, deterministic. + +## Problem + +Pick a subset S of tasks, each with `cost` (tokens), `value`, and `requires` (prerequisite ids), +such that S is **closed** under prerequisites and `sum(cost) <= budget`, maximising `sum(value)`. +This is the shape of "what work fits in my context window" for an agentic run: a step whose +prerequisites are not in the window is worthless, so a greedy scan is not safe. + +## Why the answer is exact, not heuristic + +1. `topo_order` orders tasks so every prerequisite precedes its dependents (Kahn; cycles raise). +2. DP over `(index, remaining_budget, bitmask_of_relevant_earlier_selections)`. + A bit is *relevant* at index i iff it is a prerequisite of some task at index >= i. + Restricting the mask to relevant bits is lossless: a non-relevant earlier selection can never + influence a later decision, and every prerequisite of task i is relevant at i, so the mask fully + determines whether task i may be taken. +3. The DP explores the entire include/exclude decision space; no sampling, no rounding. + +## Evidence (measured, this machine, Python 3.14.7) + +Command: `python3 /root/.hermes/wizard-outputs/null/test_budgeted_planner.py` -> exit 0, 18 tests OK. + +``` +OK + n=40 dense instance solved in 0.223s + cross-check: 300 instances, 160 with positive optimum +``` + +- **Optimality**: 300 seeded random DAG instances (n = 1..11, budgets 0..total_cost+5, edge + densities 0.0/0.15/0.3/0.5) — solver optimum == `brute_force` optimum on every instance. + `brute_force` is an independent oracle: it enumerates all 2^n subsets and tests closure by set + containment, with no topological order and no DP. +- 40 dense instances (n = 8, edge density 0.6) and 60 zero-value-task instances also cross-checked. +- Hand-computed cases: chain forces prerequisites; prerequisite blows budget; zero budget; + empty input; value-0 filler + deterministic include tie-break. +- Validation: duplicate id, dangling prerequisite, 2-cycle, self-cycle, cost <= 0, value < 0, + negative budget, non-int budget -> `SpecError` (CLI exits 2). +- Large instances (n = 60 chain, n = 60 independent) checked for closure, budget feasibility and + self-consistent accounting; greedy is a verified lower bound on value. +- Determinism: identical output across repeated runs on the same input. +- CLI smoke test: budget 9, tasks ingest(4,3), index(7,10)->ingest, report(5,6) + -> `{"cost": 9, "selected": ["ingest", "report"], "value": 9}` (index is unreachable: 4+7 > 9). + +## Limits (claimed, not measured beyond n=40) + +Exponential in the number of *relevant* earlier selections per step. Measured 0.223s at n=40 dense; +fine for tens of tasks, not for hundreds of deeply interdependent ones. `brute_force` is 2^n and +is only usable for n <= ~15. + +## Bugs the suite caught during development + +1. DP table keyed by `Task` object instead of task id -> `KeyError` in `_relevant_masks` (11 test + errors). The CLI's `except KeyError` also silently turned this into exit 2, which is why the + CLI exit-code test exists. +2. A wrong test expectation (budget 2 cannot hold tasks costing 1 and 2). + +## Usage + +``` +echo '{"budget": 9, "tasks": [{"id":"a","cost":4,"value":3},{"id":"b","cost":7,"value":10,"requires":["a"]}]}' \ + | python3 budgeted_planner.py +``` diff --git a/wizard-work/null/budgeted-planner/_test_run.txt b/wizard-work/null/budgeted-planner/_test_run.txt new file mode 100644 index 0000000..9675f95 --- /dev/null +++ b/wizard-work/null/budgeted-planner/_test_run.txt @@ -0,0 +1,27 @@ +test_cli_exit_codes (__main__.TestCliAndJson.test_cli_exit_codes) ... ok +test_json_roundtrip (__main__.TestCliAndJson.test_json_roundtrip) ... ok +test_chain_forces_prerequisites (__main__.TestHandComputed.test_chain_forces_prerequisites) ... ok +test_prereq_blows_budget (__main__.TestHandComputed.test_prereq_blows_budget) ... ok +test_tie_is_deterministic_include (__main__.TestHandComputed.test_tie_is_deterministic_include) ... ok +test_zero_budget_and_empty (__main__.TestHandComputed.test_zero_budget_and_empty) ... ok +test_zero_value_task_free_inclusion (__main__.TestHandComputed.test_zero_value_task_free_inclusion) ... ok +test_runtime_bounded (__main__.TestLargeInstances.test_runtime_bounded) ... ok +test_wide_and_deep_invariants (__main__.TestLargeInstances.test_wide_and_deep_invariants) ... ok +test_300_random_instances (__main__.TestOptimalityAgainstOracle.test_300_random_instances) ... ok +test_dense_graphs_match_oracle (__main__.TestOptimalityAgainstOracle.test_dense_graphs_match_oracle) ... ok +test_zero_value_and_zero_cost_edges (__main__.TestOptimalityAgainstOracle.test_zero_value_and_zero_cost_edges) ... ok +test_bad_cost_value_budget (__main__.TestValidation.test_bad_cost_value_budget) ... ok +test_brute_force_also_validates (__main__.TestValidation.test_brute_force_also_validates) ... ok +test_cycle (__main__.TestValidation.test_cycle) ... ok +test_dangling_prerequisite (__main__.TestValidation.test_dangling_prerequisite) ... ok +test_duplicate_id (__main__.TestValidation.test_duplicate_id) ... ok +test_self_cycle (__main__.TestValidation.test_self_cycle) ... ok + +---------------------------------------------------------------------- +Ran 18 tests in 1.585s + +OK + + n=40 dense instance solved in 0.223s + + cross-check: 300 instances, 160 with positive optimum diff --git a/wizard-work/null/budgeted-planner/budgeted_planner.py b/wizard-work/null/budgeted-planner/budgeted_planner.py new file mode 100644 index 0000000..0ebd58a --- /dev/null +++ b/wizard-work/null/budgeted-planner/budgeted_planner.py @@ -0,0 +1,277 @@ +#!/usr/bin/env python3 +"""budgeted_planner.py — exact optimal task selection under a cost budget and prerequisite closure. + +Lane: Codeforge. Pure stdlib. + +Problem +------- +Given tasks with a token `cost`, an integer `value`, and prerequisite ids `requires`, +choose a subset S that is + + * CLOSED: every prerequisite of every task in S is also in S, + * FEASIBLE: sum(cost(t) for t in S) <= budget, + +maximising sum(value(t) for t in S). + +Exactness +--------- +`topo_order` imposes an order where every prerequisite precedes its dependents. +A task's prerequisites are therefore all earlier in that order. The solver does a +memoised DP over (index, remaining budget, bitmask of *relevant* earlier selections). +A bit is "relevant" at index i iff it is the prerequisite of some task at index >= i; +restricting the mask to relevant bits is lossless because a non-relevant earlier +selection can never influence any future decision. No heuristic, no sampling: the DP +enumerates the whole decision space, so the returned value is the true optimum. + +`brute_force` is an independent oracle (enumerate all 2**n subsets, test closure by set +containment, no topo order, no DP) used by the tests as the correctness reference. + +Determinism +----------- +For a fixed input the output is byte-identical across runs. Among equally optimal +selections the solver deterministically prefers the one that includes a task when the +include/exclude values tie. + +Limits +------ +Exponential in the number of *relevant* earlier selections per step; comfortable for +tens of tasks. `brute_force` is 2**n and is only usable for n <= ~15. +""" + +from __future__ import annotations + +import json +import sys +from dataclasses import dataclass +from typing import Iterable, Mapping, Sequence + +__all__ = [ + "Task", + "SpecError", + "solve", + "brute_force", + "greedy", + "topo_order", + "check_selection", + "plan_from_json", + "main", +] + + +class SpecError(ValueError): + """Raised for malformed input: duplicate ids, dangling/cyclic prerequisites, + non-positive costs, negative values, negative budget.""" + + +@dataclass(frozen=True) +class Task: + id: str + cost: int + value: int + requires: tuple[str, ...] + + @classmethod + def of(cls, obj: Mapping) -> "Task": + return cls( + id=str(obj["id"]), + cost=int(obj["cost"]), + value=int(obj["value"]), + requires=tuple(str(r) for r in obj.get("requires", ())), + ) + + +def _validate(tasks: Sequence[Task], budget: int) -> None: + if not isinstance(budget, int) or isinstance(budget, bool): + raise SpecError(f"budget must be an int, got {type(budget).__name__}") + if budget < 0: + raise SpecError(f"budget must be >= 0, got {budget}") + seen: set[str] = set() + for t in tasks: + if not t.id: + raise SpecError("task id must be non-empty") + if t.id in seen: + raise SpecError(f"duplicate task id: {t.id!r}") + seen.add(t.id) + if t.cost <= 0: + raise SpecError(f"task {t.id!r} cost must be > 0, got {t.cost}") + if t.value < 0: + raise SpecError(f"task {t.id!r} value must be >= 0, got {t.value}") + for t in tasks: + for r in t.requires: + if r not in seen: + raise SpecError(f"task {t.id!r} requires unknown task {r!r}") + + +def topo_order(ids: Iterable[str], requires: Mapping[str, Sequence[str]]) -> list[str]: + """Return an order in which every prerequisite precedes its dependent. + + Raises SpecError if the prerequisite graph has a cycle. + """ + pending = list(ids) + emitted: list[str] = [] + done: set[str] = set() + while pending: + ready = sorted( + t for t in pending if all(r in done for r in requires.get(t, ())) + ) + if not ready: + raise SpecError( + "prerequisite cycle among: " + ", ".join(sorted(pending)) + ) + for t in ready: + emitted.append(t) + done.add(t) + pending.remove(t) + return emitted + + +def _relevant_masks(order: Sequence[str], req_idx: Mapping[str, tuple[int, ...]]) -> list[frozenset[int]]: + """relevant[i] = indices < i that are a prerequisite of some task at index >= i.""" + n = len(order) + rel: list[frozenset[int]] = [] + for i in range(n + 1): + acc: set[int] = set() + for j in range(i, n): + acc.update(req_idx[order[j]]) + rel.append(frozenset(j for j in acc if j < i)) + return rel + + +def solve(tasks: Sequence[Task], budget: int) -> dict: + """Exact optimum. Returns {'selected', 'cost', 'value'}.""" + tasks = list(tasks) + _validate(tasks, budget) + order = topo_order([t.id for t in tasks], {t.id: t.requires for t in tasks}) + idx = {t: i for i, t in enumerate(order)} + cost = {t.id: t.cost for t in tasks} + value = {t.id: t.value for t in tasks} + req_idx = {t.id: tuple(sorted(idx[r] for r in t.requires)) for t in tasks} + rel = _relevant_masks(order, req_idx) + n = len(order) + + memo: dict[tuple[int, int, int], int] = {} + + def f(i: int, rem: int, mask: int) -> int: + if i == n: + return 0 + key = (i, rem, mask) + hit = memo.get(key) + if hit is not None: + return hit + best = f(i + 1, rem, mask & _bits(rel[i + 1])) + t = order[i] + if cost[t] <= rem and all(mask >> r & 1 for r in req_idx[t]): + cand = value[t] + f(i + 1, rem - cost[t], (mask | (1 << i)) & _bits(rel[i + 1])) + if cand > best: + best = cand + memo[key] = best + return best + + chosen: list[str] = [] + + def build(i: int, rem: int, mask: int) -> None: + if i == n: + return + t = order[i] + excl = f(i + 1, rem, mask & _bits(rel[i + 1])) + if cost[t] <= rem and all(mask >> r & 1 for r in req_idx[t]): + inc = value[t] + f(i + 1, rem - cost[t], (mask | (1 << i)) & _bits(rel[i + 1])) + if inc >= excl: # deterministic tie-break: include + chosen.append(t) + build(i + 1, rem - cost[t], (mask | (1 << i)) & _bits(rel[i + 1])) + return + build(i + 1, rem, mask & _bits(rel[i + 1])) + + build(0, budget, 0) + out = sorted(chosen) + result = { + "selected": out, + "cost": sum(cost[t] for t in out), + "value": sum(value[t] for t in out), + } + if result["value"] != f(0, budget, 0): # internal consistency assertion + raise AssertionError("reconstruction diverged from DP value") + return result + + +def _bits(fs: frozenset[int]) -> int: + b = 0 + for j in fs: + b |= 1 << j + return b + + +def brute_force(tasks: Sequence[Task], budget: int) -> int: + """Independent oracle: enumerate every subset, keep closed+feasible ones, return max value.""" + tasks = list(tasks) + _validate(tasks, budget) + ids = [t.id for t in tasks] + cost = {t.id: t.cost for t in tasks} + value = {t.id: t.value for t in tasks} + req = {t.id: frozenset(t.requires) for t in tasks} + best = 0 + for mask in range(1 << len(ids)): + chosen = {ids[i] for i in range(len(ids)) if mask >> i & 1} + if not chosen: + continue + if any(not req[i] <= chosen for i in chosen): + continue + if sum(cost[i] for i in chosen) > budget: + continue + v = sum(value[i] for i in chosen) + if v > best: + best = v + return best + + +def greedy(tasks: Sequence[Task], budget: int) -> dict: + """Cheap lower bound: scan in id order, take any task whose prerequisites are + all already taken and whose cost fits. Not exact; used as a sanity floor.""" + tasks = list(tasks) + _validate(tasks, budget) + order = topo_order([t.id for t in tasks], {t.id: t.requires for t in tasks}) + by_id = {t.id: t for t in tasks} + taken: set[str] = set() + spent = 0 + gained = 0 + for tid in order: + t = by_id[tid] + if all(r in taken for r in t.requires) and spent + t.cost <= budget: + taken.add(tid) + spent += t.cost + gained += t.value + return {"selected": sorted(taken), "cost": spent, "value": gained} + + +def check_selection(tasks: Sequence[Task], budget: int, selected: Sequence[str]) -> None: + """Raise AssertionError unless `selected` is closed, feasible and internally consistent.""" + by_id = {t.id: t for t in tasks} + s = set(selected) + assert len(s) == len(selected), "selected contains duplicates" + assert s <= set(by_id), f"unknown task(s): {sorted(s - set(by_id))}" + for t in s: + missing = set(by_id[t].requires) - s + assert not missing, f"task {t!r} is missing prerequisites {sorted(missing)}" + assert sum(by_id[t].cost for t in s) <= budget, "selection exceeds budget" + + +def plan_from_json(payload: Mapping) -> dict: + tasks = [Task.of(o) for o in payload.get("tasks", ())] + return solve(tasks, int(payload.get("budget", 0))) + + +def main(argv: Sequence[str] | None = None) -> int: + argv = list(sys.argv[1:] if argv is None else argv) + raw = open(argv[0], "r", encoding="utf-8").read() if argv else sys.stdin.read() + try: + payload = json.loads(raw) + result = plan_from_json(payload) + except (json.JSONDecodeError, SpecError, KeyError) as exc: + print(json.dumps({"error": f"{type(exc).__name__}: {exc}"}), file=sys.stderr) + return 2 + print(json.dumps(result, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/wizard-work/null/budgeted-planner/test_budgeted_planner.py b/wizard-work/null/budgeted-planner/test_budgeted_planner.py new file mode 100644 index 0000000..f69afaa --- /dev/null +++ b/wizard-work/null/budgeted-planner/test_budgeted_planner.py @@ -0,0 +1,230 @@ +#!/usr/bin/env python3 +"""test_budgeted_planner.py — real test suite for budgeted_planner. + +Exits 0 iff every test passes; exits 1 on any failure (unittest semantics). +Evidence: the optimality claim is checked against an INDEPENDENT brute-force oracle +on hundreds of randomly generated instances, plus hand-computed cases, malformed +inputs, cycle detection, determinism, and a large-instance invariant sweep. + +Run: python3 /root/.hermes/wizard-outputs/null/test_budgeted_planner.py +""" + +from __future__ import annotations + +import json +import random +import subprocess +import sys +import tempfile +import time +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +from budgeted_planner import ( # noqa: E402 + SpecError, + Task, + brute_force, + check_selection, + greedy, + plan_from_json, + solve, + topo_order, +) + +MODULE = Path(__file__).resolve().parent / "budgeted_planner.py" + + +def rand_instance(rng: random.Random, n: int, edge_p: float = 0.25): + perm = [f"t{k}" for k in range(n)] + rng.shuffle(perm) # topological order + requires: dict[str, list[str]] = {t: [] for t in perm} + for a in range(n): + for b in range(a + 1, n): + if rng.random() < edge_p: + requires[perm[b]].append(perm[a]) + tasks = [ + Task( + id=t, + cost=rng.randint(1, 9), + value=rng.randint(0, 20), + requires=tuple(requires[t]), + ) + for t in perm + ] + return tasks, sum(t.cost for t in tasks) + + +class TestHandComputed(unittest.TestCase): + def test_chain_forces_prerequisites(self): + # t2 needs t1; budget 10. t2(7,10)+t1(4,3)=11 > 10 -> must drop t2. + tasks = [ + Task("t1", 4, 3, ()), + Task("t2", 7, 10, ("t1",)), + Task("t3", 5, 6, ()), + ] + r = solve(tasks, 10) + self.assertEqual(r["value"], 9) # t1 + t3 + check_selection(tasks, 10, r["selected"]) + + def test_prereq_blows_budget(self): + tasks = [Task("a", 6, 9, ()), Task("b", 5, 100, ("a",))] + r = solve(tasks, 9) + self.assertEqual(r["value"], 9) # b unreachable: 6+5 > 9 + + def test_zero_budget_and_empty(self): + self.assertEqual(solve([], 5), {"selected": [], "cost": 0, "value": 0}) + self.assertEqual(solve([Task("a", 1, 1, ())], 0)["value"], 0) + + def test_zero_value_task_free_inclusion(self): + # budget 3 fits both; value-0 task 'a' adds nothing but is taken because the + # solver's deterministic tie-break prefers include on equal value. + tasks = [Task("a", 1, 0, ()), Task("b", 2, 5, ())] + r = solve(tasks, 3) + self.assertEqual(r["value"], 5) + self.assertEqual(r["cost"], 3) + self.assertIn("a", r["selected"]) + # budget 2 cannot hold both (1+2=3): only b fits, optimum 5 either way. + self.assertEqual(solve(tasks, 2)["selected"], ["b"]) + + def test_tie_is_deterministic_include(self): + tasks = [Task("a", 5, 5, ()), Task("b", 5, 5, ())] + r1 = solve(tasks, 5) + r2 = solve(tasks, 5) + self.assertEqual(r1, r2) + self.assertEqual(len(r1["selected"]), 1) + + +class TestValidation(unittest.TestCase): + def _must_raise(self, fn, *a, **kw): + with self.assertRaises(SpecError): + fn(*a, **kw) + + def test_duplicate_id(self): + self._must_raise(solve, [Task("a", 1, 1, ()), Task("a", 1, 2, ())], 5) + + def test_dangling_prerequisite(self): + self._must_raise(solve, [Task("a", 1, 1, ("ghost",))], 5) + + def test_cycle(self): + tasks = [Task("a", 1, 1, ("b",)), Task("b", 1, 1, ("a",))] + self._must_raise(solve, tasks, 5) + with self.assertRaises(SpecError): + topo_order(["a", "b"], {"a": ["b"], "b": ["a"]}) + + def test_self_cycle(self): + self._must_raise(solve, [Task("a", 1, 1, ("a",))], 5) + + def test_bad_cost_value_budget(self): + self._must_raise(solve, [Task("a", 0, 1, ())], 5) + self._must_raise(solve, [Task("a", 1, -1, ())], 5) + self._must_raise(solve, [Task("a", 1, 1, ())], -1) + self._must_raise(solve, [Task("a", 1, 1, ())], True) + + def test_brute_force_also_validates(self): + self._must_raise(brute_force, [Task("a", 1, 1, ("x",))], 5) + + +class TestOptimalityAgainstOracle(unittest.TestCase): + """The core evidence: DP optimum == brute-force optimum on random instances.""" + + def test_300_random_instances(self): + rng = random.Random(20261023) + checked = 0 + non_trivial = 0 + for _ in range(300): + n = rng.randint(1, 11) + tasks, total_cost = rand_instance(rng, n, rng.choice([0.0, 0.15, 0.3, 0.5])) + budget = rng.choice([0, rng.randint(1, 12), rng.randint(1, total_cost + 5)]) + got = solve(tasks, budget) + want = brute_force(tasks, budget) + self.assertEqual( + got["value"], + want, + f"n={n} budget={budget} got={got['value']} brute={want} " + f"tasks={[(t.id, t.cost, t.value, t.requires) for t in tasks]}", + ) + check_selection(tasks, budget, got["selected"]) + self.assertEqual(got["value"], sum(t.value for t in tasks if t.id in set(got["selected"]))) + self.assertEqual(got["cost"], sum(t.cost for t in tasks if t.id in set(got["selected"]))) + self.assertLessEqual(greedy(tasks, budget)["value"], got["value"]) + checked += 1 + if want > 0: + non_trivial += 1 + print(f"\n cross-check: {checked} instances, {non_trivial} with positive optimum") + self.assertEqual(checked, 300) + + def test_dense_graphs_match_oracle(self): + rng = random.Random(777) + for _ in range(40): + tasks, total_cost = rand_instance(rng, 8, 0.6) + budget = rng.randint(1, total_cost) + self.assertEqual(solve(tasks, budget)["value"], brute_force(tasks, budget)) + + def test_zero_value_and_zero_cost_edges(self): + rng = random.Random(4242) + for _ in range(60): + tasks, _ = rand_instance(rng, rng.randint(1, 10), 0.35) + rebuilt = [ + Task(t.id, t.cost, 0 if rng.random() < 0.4 else t.value, t.requires) + for t in tasks + ] + budget = rng.randint(0, 30) + self.assertEqual(solve(rebuilt, budget)["value"], brute_force(rebuilt, budget)) + + +class TestLargeInstances(unittest.TestCase): + def test_wide_and_deep_invariants(self): + rng = random.Random(99) + # deep chain of 60 tasks + deep = [Task(f"c{i}", rng.randint(1, 3), rng.randint(1, 5), (f"c{i-1}",) if i else ()) for i in range(60)] + r = solve(deep, 40) + check_selection(deep, 40, r["selected"]) + self.assertEqual(r["value"], sum(t.value for t in deep if t.id in set(r["selected"]))) + self.assertGreaterEqual(r["value"], greedy(deep, 40)["value"]) + # wide: 60 independent tasks + wide = [Task(f"w{i}", rng.randint(1, 4), rng.randint(1, 9), ()) for i in range(60)] + r2 = solve(wide, 50) + check_selection(wide, 50, r2["selected"]) + self.assertLessEqual(r2["value"], sum(t.value for t in wide)) + self.assertGreaterEqual(r2["value"], greedy(wide, 50)["value"]) + + def test_runtime_bounded(self): + rng = random.Random(5) + tasks, _ = rand_instance(rng, 40, 0.2) + t0 = time.perf_counter() + solve(tasks, 60) + dt = time.perf_counter() - t0 + print(f"\n n=40 dense instance solved in {dt:.3f}s") + self.assertLess(dt, 10.0) + + +class TestCliAndJson(unittest.TestCase): + def test_json_roundtrip(self): + payload = { + "budget": 9, + "tasks": [ + {"id": "t1", "cost": 4, "value": 3, "requires": []}, + {"id": "t2", "cost": 7, "value": 10, "requires": ["t1"]}, + {"id": "t3", "cost": 5, "value": 6, "requires": []}, + ], + } + self.assertEqual(plan_from_json(payload)["value"], 9) + + def test_cli_exit_codes(self): + with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as fh: + json.dump({"budget": 5, "tasks": [{"id": "a", "cost": 5, "value": 5}]}, fh) + good = fh.name + with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as fh: + fh.write('{"budget": 5, "tasks": [{"id": "a", "cost": 5, "value": 5, "requires": ["z"]}]}') + bad = fh.name + ok = subprocess.run([sys.executable, str(MODULE), good], capture_output=True, text=True) + self.assertEqual(ok.returncode, 0) + self.assertEqual(json.loads(ok.stdout)["value"], 5) + badrun = subprocess.run([sys.executable, str(MODULE), bad], capture_output=True, text=True) + self.assertEqual(badrun.returncode, 2) + + +if __name__ == "__main__": + unittest.main(verbosity=2)