schedule() is unreliable

Usage by agent

Timeline

1. planner iteration 1 ✓Finished 3 calls · 9,363 tokens · $0.0013 · 13.0s

   Loading…

2. coder iteration 1 ✓Finished 4 calls · 27.6K tokens · $0.0030 · 18.4s

   Loading…

3. tester iteration 1 ✓Finished 3 calls · 25.4K tokens · $0.0025 · 17.4s

   Loading…

4. verify iteration 1 ✓Passed 0.5s

   ..................                                                       [100%]
   18 passed in 0.04s
   

5. reviewer iteration 1 ✓Finished 2 calls · 16.5K tokens · $0.0018 · 11.0s

   Loading…

Final diff

diff --git a/scheduler.py b/scheduler.py
index d2d5b46..71f04a8 100644
--- a/scheduler.py
+++ b/scheduler.py
@@ -1,15 +1,42 @@
+import heapq
+
 from graph import dependents
 
 
+class CycleError(ValueError):
+    """Raised when tasks contain a dependency cycle (or depend on a cycle)."""
+
+
+class UnknownTaskError(ValueError):
+    """Raised when a task depends on a name that is not a task."""
+
+
 def schedule(tasks: dict[str, list[str]]) -> list[str]:
+    for name, deps in tasks.items():
+        for dep in deps:
+            if dep not in tasks:
+                raise UnknownTaskError(
+                    f"task {name!r} depends on unknown task {dep!r}"
+                )
+
     remaining = {name: len(deps) for name, deps in tasks.items()}
+    children = dependents(tasks)
     ready = [name for name, count in remaining.items() if count == 0]
+    heapq.heapify(ready)
     order = []
     while ready:
-        name = ready.pop()
+        name = heapq.heappop(ready)
         order.append(name)
-        for child in dependents(tasks)[name]:
+        for child in children[name]:
             remaining[child] -= 1
             if remaining[child] == 0:
-                ready.append(child)
+                heapq.heappush(ready, child)
+
+    if len(order) != len(tasks):
+        done = set(order)
+        stuck = sorted(name for name in tasks if name not in done)
+        raise CycleError(
+            "dependency cycle among tasks (unschedulable, including any "
+            f"tasks downstream of a cycle): {', '.join(stuck)}"
+        )
     return order
diff --git a/test_scheduler.py b/test_scheduler.py
index 83bc850..b080223 100644
--- a/test_scheduler.py
+++ b/test_scheduler.py
@@ -1,5 +1,135 @@
-from scheduler import schedule
+import pytest
+
+from scheduler import CycleError, UnknownTaskError, schedule
 
 
 def test_chain():
     assert schedule({"a": [], "b": ["a"], "c": ["b"]}) == ["a", "b", "c"]
+
+
+def test_empty():
+    assert schedule({}) == []
+
+
+def test_cycle_raises_with_task_names():
+    with pytest.raises(CycleError) as exc:
+        schedule({"a": ["b"], "b": ["a"]})
+    assert "a" in str(exc.value) and "b" in str(exc.value)
+
+
+def test_cycle_message_lists_only_unscheduled_tasks():
+    with pytest.raises(CycleError) as exc:
+        schedule({"a": [], "b": ["c"], "c": ["b"], "d": ["a"]})
+    msg = str(exc.value)
+    assert "b" in msg and "c" in msg
+    assert "d" not in msg.split(":")[-1].split(", ")
+
+
+def test_self_dependency_is_cycle():
+    with pytest.raises(CycleError) as exc:
+        schedule({"a": ["a"]})
+    assert "a" in str(exc.value)
+
+
+def test_unknown_dependency_raises_named_error():
+    with pytest.raises(UnknownTaskError) as exc:
+        schedule({"a": ["zzz"]})
+    assert "zzz" in str(exc.value)
+
+
+def test_errors_are_value_errors():
+    assert issubclass(CycleError, ValueError)
+    assert issubclass(UnknownTaskError, ValueError)
+
+
+def test_deterministic_alphabetical_order():
+    assert schedule({"c": [], "a": [], "b": []}) == ["a", "b", "c"]
+    assert schedule({"c": [], "b": [], "a": []}) == ["a", "b", "c"]
+
+
+def test_alphabetical_among_newly_ready():
+    tasks = {"x": [], "b": ["x"], "a": ["x"]}
+    assert schedule(tasks) == ["x", "a", "b"]
+
+
+def test_duplicate_dependencies():
+    assert schedule({"a": [], "b": ["a", "a"]}) == ["a", "b"]
+
+
+def test_cycle_message_names_only_cycle_and_downstream_tasks():
+    tasks = {"alpha": ["beta"], "beta": ["alpha"], "gamma": [], "delta": ["gamma"]}
+    with pytest.raises(CycleError) as exc:
+        schedule(tasks)
+    msg = str(exc.value)
+    assert "alpha" in msg and "beta" in msg
+    assert "gamma" not in msg and "delta" not in msg
+
+
+def test_cycle_with_ready_tasks_still_raises_no_partial_order():
+    tasks = {"ready1": [], "ready2": ["ready1"], "xx": ["yy"], "yy": ["xx"]}
+    with pytest.raises(CycleError) as exc:
+        schedule(tasks)
+    msg = str(exc.value)
+    assert "xx" in msg and "yy" in msg
+    assert "ready1" not in msg and "ready2" not in msg
+
+
+def test_cycle_error_message_lists_stuck_names_sorted():
+    with pytest.raises(CycleError) as exc:
+        schedule({"zed": ["amy"], "amy": ["zed"], "kim": ["zed"]})
+    tail = str(exc.value).split(": ")[-1]
+    assert tail == "amy, kim, zed"
+
+
+def test_unknown_dependency_message_names_declaring_task():
+    with pytest.raises(UnknownTaskError) as exc:
+        schedule({"build": [], "deploy": ["build", "missing_dep"]})
+    msg = str(exc.value)
+    assert "missing_dep" in msg
+    assert "deploy" in msg
+
+
+def test_unknown_dependency_is_not_key_error():
+    with pytest.raises(UnknownTaskError):
+        schedule({"a": ["zzz"]})
+    try:
+        schedule({"a": ["zzz"]})
+    except KeyError:
+        pytest.fail("bare KeyError escaped")
+    except UnknownTaskError:
+        pass
+
+
+def test_alphabetical_after_x_completes_with_unrelated_ready_task():
+    tasks = {"x": [], "b": ["x"], "a": ["x"], "m": []}
+    assert schedule(tasks) == ["m", "x", "a", "b"]
+
+
+def test_duplicate_dependency_with_cycle_still_detected():
+    with pytest.raises(CycleError):
+        schedule({"a": ["b", "b"], "b": ["a"]})
+
+
+def test_random_dags_follow_smallest_ready_rule():
+    import random
+
+    rng = random.Random(1234)
+    for _ in range(200):
+        names = [f"t{i:02d}" for i in range(rng.randint(1, 12))]
+        rng.shuffle(names)
+        tasks = {}
+        for i, name in enumerate(names):
+            earlier = names[:i]
+            deps = [d for d in earlier if rng.random() < 0.3]
+            if deps and rng.random() < 0.3:
+                deps.append(rng.choice(deps))  # duplicate entry
+            tasks[name] = deps
+        order = schedule(tasks)
+        assert sorted(order) == sorted(tasks)
+
+        # Replay: at every step the chosen task must be the smallest ready one.
+        done = set()
+        for name in order:
+            ready = [t for t in tasks if t not in done and set(tasks[t]) <= done]
+            assert name == min(ready)
+            done.add(name)