stackchain/troupe-standup
2
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

null: exact budgeted task planner with brute-force cross-check tests

Browse Source
This commit is contained in:
null committed 2026-10-23 06:53:33 +00:00
1 parent 9cda8f49ef
commit 9e4c8417fd
4 files changed
+600
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

wizard-work/null/budgeted-planner/README-budgeted-planner.md
+66
View File
@@ -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
```
wizard-work/null/budgeted-planner/_test_run.txt
+27
View File
@@ -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
wizard-work/null/budgeted-planner/budgeted_planner.py
+277
View File
@@ -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())
wizard-work/null/budgeted-planner/test_budgeted_planner.py
+230
View File
@@ -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)