fable-5.1/01-feature-building/RESPONSE.md

cd /home/user/agent-eval-fable-5.1/01-feature-building && python3 -B -m unittest -v test_jobqueue

----------------------------------------------------------------------
Ran 36 tests in 0.065s

OK

fable-5.1/01-feature-building/TASK.md

fable-5.1/01-feature-building/jobqueue.py

"""Dependency-aware job scheduling.

`plan_jobs` turns a collection of `Job` records into a deterministic execution
order that respects every dependency, prefers higher-priority work among the
jobs that are ready, and rejects malformed or cyclic input with a
`ScheduleError` whose message says what went wrong.
"""

from __future__ import annotations

import heapq
from dataclasses import dataclass, field
from typing import Any, Iterable


class ScheduleError(ValueError):
    """Raised when a set of jobs cannot be turned into an execution plan."""


@dataclass(frozen=True)
class Job:
    id: str
    priority: int = 0
    depends_on: tuple[str, ...] = field(default_factory=tuple)
    payload: Any = None


def plan_jobs(jobs: Iterable[Job]) -> list[str]:
    """Return job IDs in a valid deterministic execution order.

    Guarantees:

    * Every job ID appears exactly once, and only after all of its
      dependencies.
    * Whenever several jobs are ready (all dependencies already placed), the
      one with the highest ``priority`` goes first; equal priorities are
      broken by the job's position in the input.  Readiness is re-evaluated
      after every placement, so a high-priority job that has just been
      unblocked jumps ahead of lower-priority jobs that were ready earlier.
      There is no priority inheritance: a low-priority dependency is not
      promoted because something important is waiting on it.
    * The input jobs and their ``depends_on`` collections are never modified.
    * Empty input yields ``[]``.

    Raises ``ScheduleError`` when a job ID is not a non-empty string, when
    two jobs share an ID, when a dependency is not a string or does not name
    a supplied job, when ``depends_on`` is not a collection of IDs, or when
    the dependency graph contains a cycle.  Cycle errors spell out one
    concrete cycle, e.g. ``"Dependency cycle detected: a -> b -> a"``.
    """
    job_list = list(jobs)
    if not job_list:
        return []

    index_of = _validate_ids(job_list)
    deps_of = _resolve_dependencies(job_list, index_of)

    count = len(job_list)
    dependents: list[list[int]] = [[] for _ in range(count)]
    unmet: list[int] = [0] * count
    for job_index, dep_indexes in enumerate(deps_of):
        unmet[job_index] = len(dep_indexes)
        for dep_index in dep_indexes:
            dependents[dep_index].append(job_index)

    # Kahn's algorithm with a heap so that, among the ready jobs, the highest
    # priority wins and ties fall back to input order.  The input index is
    # unique, so heap entries never need to compare Job objects.
    ready: list[tuple[int, int]] = [
        (-job_list[i].priority, i) for i in range(count) if unmet[i] == 0
    ]
    heapq.heapify(ready)

    order: list[str] = []
    while ready:
        _, job_index = heapq.heappop(ready)
        order.append(job_list[job_index].id)
        for dependent in dependents[job_index]:
            unmet[dependent] -= 1
            if unmet[dependent] == 0:
                heapq.heappush(ready, (-job_list[dependent].priority, dependent))

    if len(order) != count:
        cycle = _find_cycle(job_list, deps_of, unmet)
        raise ScheduleError("Dependency cycle detected: " + " -> ".join(cycle))

    return order


def _validate_ids(job_list: list[Job]) -> dict[str, int]:
    """Check that every ID is a unique, non-empty string; map ID -> index."""
    index_of: dict[str, int] = {}
    for position, job in enumerate(job_list):
        job_id = job.id
        if not isinstance(job_id, str):
            raise ScheduleError(
                f"Job at position {position} has a non-string id: {job_id!r}"
            )
        if not job_id:
            raise ScheduleError(f"Job at position {position} has an empty id")
        if job_id in index_of:
            raise ScheduleError(
                f"Duplicate job id {job_id!r} at positions "
                f"{index_of[job_id]} and {position}"
            )
        index_of[job_id] = position
    return index_of


def _resolve_dependencies(
    job_list: list[Job], index_of: dict[str, int]
) -> list[tuple[int, ...]]:
    """Translate each job's dependency IDs into de-duplicated job indexes.

    Only reads from the supplied jobs; the returned structure is fresh, so
    the caller's ``depends_on`` collections are never touched.
    """
    deps_of: list[tuple[int, ...]] = []
    for job in job_list:
        raw = job.depends_on
        if isinstance(raw, (str, bytes)):
            raise ScheduleError(
                f"Job {job.id!r}: depends_on must be a collection of job ids, "
                f"not the bare string {raw!r}"
            )
        try:
            dep_ids = list(raw)
        except TypeError:
            raise ScheduleError(
                f"Job {job.id!r}: depends_on must be an iterable of job ids, "
                f"got {raw!r}"
            ) from None

        resolved: list[int] = []
        seen: set[int] = set()
        unknown: list[str] = []
        for dep in dep_ids:
            if not isinstance(dep, str):
                raise ScheduleError(
                    f"Job {job.id!r} has a non-string dependency: {dep!r}"
                )
            dep_index = index_of.get(dep)
            if dep_index is None:
                if dep not in unknown:
                    unknown.append(dep)
                continue
            if dep_index not in seen:
                seen.add(dep_index)
                resolved.append(dep_index)
        if unknown:
            listed = ", ".join(repr(dep) for dep in unknown)
            noun = "job" if len(unknown) == 1 else "jobs"
            raise ScheduleError(f"Job {job.id!r} depends on unknown {noun}: {listed}")
        deps_of.append(tuple(resolved))
    return deps_of


def _find_cycle(
    job_list: list[Job], deps_of: list[tuple[int, ...]], unmet: list[int]
) -> list[str]:
    """Return the IDs of one concrete dependency cycle, closed on itself.

    Called only after Kahn's algorithm stalls.  Every job that was never
    placed still has at least one never-placed dependency, so walking from
    any stuck job along stuck dependencies must eventually revisit a job;
    the walk from that first visit is a cycle.  The walk is iterative, so
    very long cycles do not hit the recursion limit, and it always starts at
    the earliest stuck job and follows the first stuck dependency, which
    keeps the diagnostic deterministic.
    """
    stuck = {i for i, pending in enumerate(unmet) if pending > 0}
    path: list[int] = []
    first_seen_at: dict[int, int] = {}
    current = min(stuck)
    while current not in first_seen_at:
        first_seen_at[current] = len(path)
        path.append(current)
        current = next(dep for dep in deps_of[current] if dep in stuck)
    loop = path[first_seen_at[current]:] + [current]
    return [job_list[i].id for i in loop]

fable-5.1/01-feature-building/test_jobqueue.py

import random
import unittest

from jobqueue import Job, ScheduleError, plan_jobs


def reference_plan(jobs):
    """Slow but obviously-correct scheduler used to cross-check plan_jobs.

    Repeatedly picks, among the jobs whose dependencies are all placed, the
    one with the highest priority, breaking ties by input position.
    """
    jobs = list(jobs)
    placed = []
    remaining = list(range(len(jobs)))
    while remaining:
        ready = [
            i
            for i in remaining
            if all(dep in placed for dep in jobs[i].depends_on)
        ]
        assert ready, "reference scheduler stalled on a cycle"
        best = max(ready, key=lambda i: (jobs[i].priority, -i))
        placed.append(jobs[best].id)
        remaining.remove(best)
    return placed


def random_dag(rng, size):
    """Build a random acyclic job set with random priorities and input order."""
    ids = [f"job{n}" for n in range(size)]
    hidden_order = ids[:]
    rng.shuffle(hidden_order)
    jobs = []
    for position, job_id in enumerate(hidden_order):
        candidates = hidden_order[:position]
        dep_count = min(len(candidates), rng.choice([0, 0, 1, 1, 2, 3]))
        deps = tuple(rng.sample(candidates, dep_count))
        jobs.append(Job(job_id, priority=rng.randint(-3, 3), depends_on=deps))
    rng.shuffle(jobs)
    return jobs


def assert_valid_plan(test, jobs, plan):
    """Every ID exactly once, every dependency before its dependant."""
    jobs = list(jobs)
    test.assertEqual(sorted(plan), sorted(job.id for job in jobs))
    test.assertEqual(len(plan), len(set(plan)))
    position = {job_id: index for index, job_id in enumerate(plan)}
    for job in jobs:
        for dep in job.depends_on:
            test.assertLess(
                position[dep], position[job.id], f"{dep} must run before {job.id}"
            )


class PlanJobsOrderingTests(unittest.TestCase):
    def test_dependencies_override_priority(self):
        jobs = [
            Job("build", priority=100, depends_on=("fetch",)),
            Job("fetch", priority=1),
        ]
        self.assertEqual(plan_jobs(jobs), ["fetch", "build"])

    def test_ready_jobs_use_priority_then_input_order(self):
        jobs = [Job("a", 2), Job("b", 5), Job("c", 2)]
        self.assertEqual(plan_jobs(jobs), ["b", "a", "c"])

    def test_empty_input_returns_empty_list(self):
        self.assertEqual(plan_jobs([]), [])
        self.assertEqual(plan_jobs(()), [])
        self.assertEqual(plan_jobs(iter([])), [])

    def test_single_job(self):
        self.assertEqual(plan_jobs([Job("only")]), ["only"])

    def test_accepts_a_one_shot_iterator(self):
        jobs = (job for job in [Job("b", depends_on=("a",)), Job("a")])
        self.assertEqual(plan_jobs(jobs), ["a", "b"])

    def test_list_dependencies_are_accepted(self):
        jobs = [Job("c", depends_on=["a", "b"]), Job("a"), Job("b")]
        self.assertEqual(plan_jobs(jobs), ["a", "b", "c"])

    def test_priority_sort_is_stable_across_many_ready_jobs(self):
        jobs = [Job("p3a", 3), Job("p1a", 1), Job("p3b", 3), Job("p2", 2), Job("p1b", 1)]
        self.assertEqual(plan_jobs(jobs), ["p3a", "p3b", "p2", "p1a", "p1b"])

    def test_negative_priorities_sort_below_zero(self):
        jobs = [Job("neg", -5), Job("zero"), Job("pos", 5)]
        self.assertEqual(plan_jobs(jobs), ["pos", "zero", "neg"])

    def test_newly_unblocked_high_priority_job_jumps_ahead(self):
        # Once "gate" finishes, "high" becomes ready and must beat "mid",
        # which had been ready from the start.
        jobs = [Job("gate", 5), Job("mid", 3), Job("high", 10, depends_on=("gate",))]
        self.assertEqual(plan_jobs(jobs), ["gate", "high", "mid"])

    def test_no_priority_inheritance_through_dependencies(self):
        # A high-priority job waiting on a low-priority gate does not pull
        # the gate forward: only ready jobs are compared.
        jobs = [Job("low", 1), Job("gate", 0), Job("high", 10, depends_on=("gate",))]
        self.assertEqual(plan_jobs(jobs), ["low", "gate", "high"])

    def test_equal_priority_ties_use_input_order_not_readiness_time(self):
        # "c" is unblocked after "d" was already ready, but "c" comes earlier
        # in the input, so it still wins the tie.
        jobs = [Job("a"), Job("x"), Job("c", depends_on=("x",)), Job("d")]
        self.assertEqual(plan_jobs(jobs), ["a", "x", "c", "d"])

    def test_diamond_dependencies(self):
        jobs = [
            Job("top", priority=9, depends_on=("left", "right")),
            Job("left", priority=1, depends_on=("root",)),
            Job("right", priority=2, depends_on=("root",)),
            Job("root"),
        ]
        self.assertEqual(plan_jobs(jobs), ["root", "right", "left", "top"])

    def test_duplicate_dependencies_within_one_job_are_tolerated(self):
        jobs = [Job("b", depends_on=("a", "a", "a")), Job("a")]
        self.assertEqual(plan_jobs(jobs), ["a", "b"])

    def test_disconnected_components_interleave_by_priority(self):
        jobs = [
            Job("x2", 1, depends_on=("x1",)),
            Job("y1", 5),
            Job("x1", 2),
            Job("y2", 0, depends_on=("y1",)),
        ]
        self.assertEqual(plan_jobs(jobs), ["y1", "x1", "x2", "y2"])

    def test_random_dags_match_reference_scheduler(self):
        rng = random.Random(20260929)
        for size in (1, 2, 5, 10, 25, 60):
            for _ in range(20):
                jobs = random_dag(rng, size)
                plan = plan_jobs(jobs)
                assert_valid_plan(self, jobs, plan)
                self.assertEqual(plan, reference_plan(jobs))

    def test_result_is_deterministic(self):
        jobs = random_dag(random.Random(7), 40)
        first = plan_jobs(jobs)
        self.assertEqual(first, plan_jobs(jobs))
        self.assertEqual(first, plan_jobs(list(jobs)))

    def test_long_chain_does_not_hit_recursion_limit(self):
        size = 5000
        jobs = [Job(f"n{i}", depends_on=(f"n{i - 1}",)) for i in range(1, size)]
        jobs.append(Job("n0"))
        plan = plan_jobs(jobs)
        self.assertEqual(plan, [f"n{i}" for i in range(size)])


class PlanJobsValidationTests(unittest.TestCase):
    def test_unknown_dependency_is_rejected(self):
        with self.assertRaises(ScheduleError):
            plan_jobs([Job("build", depends_on=("missing",))])

    def test_unknown_dependency_message_names_job_and_dependency(self):
        with self.assertRaisesRegex(ScheduleError, r"'build'.*'missing'"):
            plan_jobs([Job("build", depends_on=("missing",))])

    def test_all_unknown_dependencies_of_a_job_are_listed(self):
        with self.assertRaisesRegex(ScheduleError, r"'nope1'.*'nope2'"):
            plan_jobs([Job("a"), Job("b", depends_on=("a", "nope1", "nope2"))])

    def test_duplicate_id_is_rejected(self):
        with self.assertRaisesRegex(ScheduleError, r"Duplicate job id 'dup'"):
            plan_jobs([Job("dup"), Job("other"), Job("dup")])

    def test_empty_id_is_rejected(self):
        with self.assertRaisesRegex(ScheduleError, r"empty id"):
            plan_jobs([Job("")])

    def test_non_string_id_is_rejected(self):
        for bad in (None, 7, b"bytes"):
            with self.subTest(bad=bad):
                with self.assertRaisesRegex(ScheduleError, r"non-string id"):
                    plan_jobs([Job(bad)])

    def test_non_string_dependency_is_rejected(self):
        with self.assertRaisesRegex(ScheduleError, r"'b'.*non-string dependency"):
            plan_jobs([Job("a"), Job("b", depends_on=("a", 3))])

    def test_bare_string_depends_on_is_rejected(self):
        # A common slip: depends_on="a" instead of depends_on=("a",).
        with self.assertRaisesRegex(ScheduleError, r"bare string"):
            plan_jobs([Job("a"), Job("b", depends_on="a")])

    def test_non_iterable_depends_on_is_rejected(self):
        with self.assertRaisesRegex(ScheduleError, r"depends_on must be an iterable"):
            plan_jobs([Job("a", depends_on=None)])

    def test_validation_happens_before_any_scheduling(self):
        # An invalid job at the end of the input must still be reported.
        jobs = [Job("a"), Job("b", depends_on=("a",)), Job("")]
        with self.assertRaises(ScheduleError):
            plan_jobs(jobs)

    def test_schedule_error_is_a_value_error(self):
        with self.assertRaises(ValueError):
            plan_jobs([Job("a", depends_on=("a",))])


class PlanJobsCycleTests(unittest.TestCase):
    def test_self_dependency_is_a_cycle(self):
        with self.assertRaisesRegex(ScheduleError, r"cycle.*\ba -> a\b"):
            plan_jobs([Job("a", depends_on=("a",))])

    def test_two_node_cycle_names_both_jobs(self):
        with self.assertRaisesRegex(ScheduleError, r"cycle.*\ba -> b -> a\b"):
            plan_jobs([Job("a", depends_on=("b",)), Job("b", depends_on=("a",))])

    def test_longer_cycle_lists_only_the_cycle_members(self):
        jobs = [
            Job("start"),
            Job("downstream", depends_on=("c",)),
            Job("a", depends_on=("c",)),
            Job("b", depends_on=("a",)),
            Job("c", depends_on=("b",)),
            Job("leaf", depends_on=("start",)),
        ]
        with self.assertRaises(ScheduleError) as caught:
            plan_jobs(jobs)
        message = str(caught.exception)
        self.assertIn("Dependency cycle detected", message)
        self.assertIn("c -> b -> a -> c", message)
        for outsider in ("start", "downstream", "leaf"):
            self.assertNotIn(outsider, message)

    def test_cycle_is_detected_even_when_priorities_are_high(self):
        jobs = [Job("x", 100, depends_on=("y",)), Job("y", 100, depends_on=("x",))]
        with self.assertRaises(ScheduleError):
            plan_jobs(jobs)

    def test_cycle_among_multiple_cycles_reports_one_concrete_cycle(self):
        jobs = [
            Job("p", depends_on=("q",)),
            Job("q", depends_on=("p",)),
            Job("r", depends_on=("s",)),
            Job("s", depends_on=("r",)),
        ]
        with self.assertRaisesRegex(ScheduleError, r"p -> q -> p"):
            plan_jobs(jobs)

    def test_very_long_cycle_does_not_hit_recursion_limit(self):
        size = 5000
        jobs = [Job(f"n{i}", depends_on=(f"n{(i + 1) % size}",)) for i in range(size)]
        with self.assertRaises(ScheduleError) as caught:
            plan_jobs(jobs)
        self.assertIn("n0", str(caught.exception))
        self.assertIn(f"n{size - 1}", str(caught.exception))


class PlanJobsImmutabilityTests(unittest.TestCase):
    def test_inputs_are_not_mutated(self):
        deps_c = ["a", "b", "a"]
        deps_b = ["a"]
        jobs = [
            Job("c", 1, depends_on=deps_c, payload={"k": "v"}),
            Job("b", 2, depends_on=deps_b),
            Job("a", 3),
        ]
        original_jobs = list(jobs)
        snapshot = [(job.id, job.priority, list(job.depends_on), job.payload) for job in jobs]

        self.assertEqual(plan_jobs(jobs), ["a", "b", "c"])

        self.assertEqual(jobs, original_jobs)
        self.assertIs(jobs[0].depends_on, deps_c)
        self.assertEqual(deps_c, ["a", "b", "a"])
        self.assertEqual(deps_b, ["a"])
        self.assertEqual(
            [(job.id, job.priority, list(job.depends_on), job.payload) for job in jobs],
            snapshot,
        )

    def test_returns_a_fresh_list_each_call(self):
        jobs = [Job("a"), Job("b")]
        first = plan_jobs(jobs)
        second = plan_jobs(jobs)
        self.assertEqual(first, second)
        self.assertIsNot(first, second)
        first.append("tampered")
        self.assertEqual(plan_jobs(jobs), ["a", "b"])


if __name__ == "__main__":
    unittest.main()

sonnet-5.5/01-feature-building/RESPONSE.md

python3 -m unittest -v

Ran 43 tests in 0.034s

OK

sonnet-5.5/01-feature-building/TASK.md

sonnet-5.5/01-feature-building/jobqueue.py

import heapq
import numbers
from dataclasses import dataclass, field
from typing import Any, Iterable


class ScheduleError(ValueError):
    pass


@dataclass(frozen=True)
class Job:
    id: str
    priority: int = 0
    depends_on: tuple[str, ...] = field(default_factory=tuple)
    payload: Any = None


def _validate(jobs: Iterable[Job]) -> tuple[list[Job], list[tuple[str, ...]]]:
    """Materialise ``jobs`` once and check every input rule.

    Returns the jobs in input order together with a normalised (tuple) copy of
    each job's dependencies, so ``depends_on`` is read exactly once even if the
    caller supplied a one-shot iterable.
    """
    job_list = list(jobs)
    seen: set[str] = set()
    for position, job in enumerate(job_list):
        if not isinstance(job, Job):
            raise ScheduleError(
                f"item at position {position} is not a Job: {job!r}"
            )
        if not isinstance(job.id, str) or not job.id:
            raise ScheduleError(
                f"job at position {position} must have a non-empty string id, "
                f"got {job.id!r}"
            )
        if job.id in seen:
            raise ScheduleError(f"duplicate job id: {job.id!r}")
        seen.add(job.id)
        if (
            not isinstance(job.priority, numbers.Real)
            or job.priority != job.priority  # NaN has no defined ordering
        ):
            raise ScheduleError(
                f"job {job.id!r} has an invalid priority: {job.priority!r}"
            )

    deps_by_job: list[tuple[str, ...]] = []
    for job in job_list:
        raw = job.depends_on
        if isinstance(raw, (str, bytes)):
            # A bare string would silently be iterated character by character.
            raise ScheduleError(
                f"job {job.id!r}: depends_on must be a collection of job ids, "
                f"not a single string ({raw!r})"
            )
        try:
            deps = tuple(raw)
        except TypeError:
            raise ScheduleError(
                f"job {job.id!r}: depends_on must be iterable, got {raw!r}"
            ) from None
        for dep in deps:
            if not isinstance(dep, str):
                raise ScheduleError(
                    f"job {job.id!r} has a non-string dependency: {dep!r}"
                )
            if dep not in seen:
                raise ScheduleError(
                    f"job {job.id!r} depends on unknown job {dep!r}"
                )
        deps_by_job.append(deps)
    return job_list, deps_by_job


def _find_cycle(
    stuck: set[int],
    deps_by_job: list[tuple[str, ...]],
    index_of: dict[str, int],
) -> list[int]:
    """Return the indices of one dependency cycle among the ``stuck`` jobs.

    Every stuck job still has at least one unfinished dependency, and that
    dependency is itself stuck.  Walking "depends on" edges from any stuck job
    therefore has to revisit a job, which closes a cycle.  The walk starts at
    the earliest stuck job and follows the earliest listed stuck dependency, so
    the reported cycle is deterministic.
    """
    start = min(stuck)
    path: list[int] = []
    position: dict[int, int] = {}
    node = start
    while node not in position:
        position[node] = len(path)
        path.append(node)
        node = next(
            index_of[dep] for dep in deps_by_job[node] if index_of[dep] in stuck
        )
    return path[position[node]:]


def plan_jobs(jobs: Iterable[Job]) -> list[str]:
    """Return job IDs in a valid deterministic execution order.

    Every job appears exactly once, after all of its dependencies.  Whenever
    several jobs are ready at the same time the one with the highest
    ``priority`` runs first, and equal priorities fall back to input order.

    Raises ``ScheduleError`` for malformed input (non-Job items, empty or
    non-string or duplicate ids, unknown dependencies, invalid priorities) and
    for dependency cycles.  The supplied jobs are never modified.
    """
    job_list, deps_by_job = _validate(jobs)
    index_of = {job.id: i for i, job in enumerate(job_list)}

    # Kahn's algorithm.  Duplicate entries in one depends_on collapse, so a job
    # is counted once per distinct dependency.
    unmet = [0] * len(job_list)
    dependents: list[list[int]] = [[] for _ in job_list]
    for i, deps in enumerate(deps_by_job):
        for dep in set(deps):
            unmet[i] += 1
            dependents[index_of[dep]].append(i)

    # heapq is a min-heap: negate priority so the highest priority pops first
    # and use the input index as the tie-breaker.
    ready = [(-job_list[i].priority, i) for i in range(len(job_list)) if not unmet[i]]
    heapq.heapify(ready)

    order: list[str] = []
    while ready:
        _, i = heapq.heappop(ready)
        order.append(job_list[i].id)
        for child in dependents[i]:
            unmet[child] -= 1
            if not unmet[child]:
                heapq.heappush(ready, (-job_list[child].priority, child))

    if len(order) != len(job_list):
        stuck = {i for i, count in enumerate(unmet) if count}
        cycle = _find_cycle(stuck, deps_by_job, index_of)
        ids = [job_list[i].id for i in cycle]
        chain = " -> ".join(repr(x) for x in [*ids, ids[0]])
        blocked = len(stuck) - len(cycle)
        message = f"dependency cycle detected: {chain}"
        if blocked:
            message += f" ({blocked} more job(s) are blocked behind it)"
        raise ScheduleError(message)

    return order

sonnet-5.5/01-feature-building/test_jobqueue.py

import copy
import unittest

from jobqueue import Job, ScheduleError, plan_jobs


def assert_valid_order(test, jobs, order):
    """Every id exactly once, and every job after all of its dependencies."""
    test.assertEqual(sorted(order), sorted(j.id for j in jobs))
    position = {job_id: i for i, job_id in enumerate(order)}
    for job in jobs:
        for dep in job.depends_on:
            test.assertLess(position[dep], position[job.id], f"{dep} before {job.id}")


class PlanJobsTests(unittest.TestCase):
    def test_dependencies_override_priority(self):
        jobs = [
            Job("build", priority=100, depends_on=("fetch",)),
            Job("fetch", priority=1),
        ]
        self.assertEqual(plan_jobs(jobs), ["fetch", "build"])

    def test_ready_jobs_use_priority_then_input_order(self):
        jobs = [Job("a", 2), Job("b", 5), Job("c", 2)]
        self.assertEqual(plan_jobs(jobs), ["b", "a", "c"])

    def test_unknown_dependency_is_rejected(self):
        with self.assertRaises(ScheduleError):
            plan_jobs([Job("build", depends_on=("missing",))])

    # -- ordering ---------------------------------------------------------

    def test_empty_input_returns_empty_list(self):
        self.assertEqual(plan_jobs([]), [])
        self.assertEqual(plan_jobs(iter(())), [])

    def test_single_job(self):
        self.assertEqual(plan_jobs([Job("only")]), ["only"])

    def test_equal_priorities_keep_input_order(self):
        jobs = [Job(name) for name in ("z", "m", "a", "q")]
        self.assertEqual(plan_jobs(jobs), ["z", "m", "a", "q"])

    def test_negative_and_mixed_priorities(self):
        jobs = [Job("low", -5), Job("zero", 0), Job("high", 3), Job("mid", -1)]
        self.assertEqual(plan_jobs(jobs), ["high", "zero", "mid", "low"])

    def test_float_priorities_are_supported(self):
        jobs = [Job("a", 0.5), Job("b", 1.5), Job("c", 1.5)]
        self.assertEqual(plan_jobs(jobs), ["b", "c", "a"])

    def test_priority_only_competes_among_currently_ready_jobs(self):
        # "urgent" has the top priority but must wait for "gate", so the
        # already-ready "medium" job goes first, then "gate", then "urgent".
        jobs = [
            Job("urgent", 100, depends_on=("gate",)),
            Job("gate", 1),
            Job("medium", 50),
        ]
        self.assertEqual(plan_jobs(jobs), ["medium", "gate", "urgent"])

    def test_newly_unlocked_job_competes_with_still_ready_jobs(self):
        jobs = [
            Job("first", 9),
            Job("unlocked", 10, depends_on=("first",)),
            Job("waiting", 5),
        ]
        self.assertEqual(plan_jobs(jobs), ["first", "unlocked", "waiting"])

    def test_unlocked_job_with_low_priority_waits_for_higher_ready_jobs(self):
        jobs = [
            Job("first", 9),
            Job("unlocked", 1, depends_on=("first",)),
            Job("waiting", 5),
        ]
        self.assertEqual(plan_jobs(jobs), ["first", "waiting", "unlocked"])

    def test_dependents_tie_break_by_original_input_order(self):
        # After "root" finishes, x and y unlock together with equal priority.
        jobs = [
            Job("y", 1, depends_on=("root",)),
            Job("x", 1, depends_on=("root",)),
            Job("root", 0),
        ]
        self.assertEqual(plan_jobs(jobs), ["root", "y", "x"])

    def test_diamond(self):
        jobs = [
            Job("top", 0, depends_on=("left", "right")),
            Job("left", 1, depends_on=("base",)),
            Job("right", 2, depends_on=("base",)),
            Job("base", 0),
        ]
        self.assertEqual(plan_jobs(jobs), ["base", "right", "left", "top"])

    def test_job_needing_several_dependencies_waits_for_all(self):
        jobs = [
            Job("c", 99, depends_on=("a", "b")),
            Job("a", 1),
            Job("b", 0),
        ]
        self.assertEqual(plan_jobs(jobs), ["a", "b", "c"])

    def test_duplicate_entries_in_depends_on_are_harmless(self):
        jobs = [Job("b", depends_on=("a", "a", "a")), Job("a")]
        self.assertEqual(plan_jobs(jobs), ["a", "b"])

    def test_dependencies_declared_as_list_are_accepted(self):
        jobs = [Job("b", depends_on=["a"]), Job("a")]
        self.assertEqual(plan_jobs(jobs), ["a", "b"])

    def test_accepts_one_shot_iterable(self):
        jobs = [Job("b", depends_on=("a",)), Job("a")]
        self.assertEqual(plan_jobs(iter(jobs)), ["a", "b"])
        self.assertEqual(plan_jobs(job for job in jobs), ["a", "b"])

    def test_disconnected_components_interleave_by_priority(self):
        jobs = [
            Job("a1", 1),
            Job("a2", 1, depends_on=("a1",)),
            Job("b1", 5),
            Job("b2", 0, depends_on=("b1",)),
        ]
        self.assertEqual(plan_jobs(jobs), ["b1", "a1", "a2", "b2"])

    def test_deterministic_and_repeatable(self):
        jobs = [
            Job("d", 1, depends_on=("b", "c")),
            Job("c", 1, depends_on=("a",)),
            Job("b", 1, depends_on=("a",)),
            Job("a", 1),
        ]
        first = plan_jobs(jobs)
        for _ in range(5):
            self.assertEqual(plan_jobs(jobs), first)
        self.assertEqual(first, ["a", "c", "b", "d"])

    def test_long_chain_does_not_hit_recursion_limit(self):
        n = 5000
        jobs = [Job("j0")] + [
            Job(f"j{i}", depends_on=(f"j{i - 1}",)) for i in range(1, n)
        ]
        # Reverse input order so dependencies always come later in the list.
        jobs.reverse()
        order = plan_jobs(jobs)
        self.assertEqual(order, [f"j{i}" for i in range(n)])

    def test_wide_fan_out_is_valid_and_sorted_by_priority(self):
        jobs = [Job("root")] + [
            Job(f"leaf{i}", priority=i % 7, depends_on=("root",)) for i in range(200)
        ]
        order = plan_jobs(jobs)
        assert_valid_order(self, jobs, order)
        priorities = {j.id: j.priority for j in jobs}
        leaf_priorities = [priorities[x] for x in order[1:]]
        self.assertEqual(leaf_priorities, sorted(leaf_priorities, reverse=True))

    def test_order_is_valid_for_a_scrambled_dag(self):
        jobs = [
            Job("e", 3, depends_on=("c", "d")),
            Job("a", 1),
            Job("d", 8, depends_on=("a",)),
            Job("c", 2, depends_on=("a", "b")),
            Job("b", 5),
            Job("f", 9, depends_on=("e",)),
        ]
        order = plan_jobs(jobs)
        assert_valid_order(self, jobs, order)
        self.assertEqual(order, ["b", "a", "d", "c", "e", "f"])

    # -- input validation -------------------------------------------------

    def test_duplicate_ids_are_rejected(self):
        with self.assertRaisesRegex(ScheduleError, "duplicate.*'a'"):
            plan_jobs([Job("a"), Job("b"), Job("a", priority=5)])

    def test_empty_id_is_rejected(self):
        with self.assertRaisesRegex(ScheduleError, "non-empty"):
            plan_jobs([Job("")])

    def test_non_string_id_is_rejected(self):
        for bad in (None, 7, b"a", ("a",)):
            with self.subTest(bad=bad):
                with self.assertRaisesRegex(ScheduleError, "non-empty string"):
                    plan_jobs([Job(bad)])

    def test_unknown_dependency_message_names_both_jobs(self):
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs([Job("a"), Job("build", depends_on=("a", "missing"))])
        message = str(ctx.exception)
        self.assertIn("build", message)
        self.assertIn("missing", message)

    def test_dependency_ids_are_case_sensitive(self):
        with self.assertRaises(ScheduleError):
            plan_jobs([Job("a"), Job("b", depends_on=("A",))])

    def test_bare_string_depends_on_is_rejected(self):
        # ("ab") is just "ab"; iterating it would look like jobs "a" and "b".
        jobs = [Job("a"), Job("b"), Job("c", depends_on="ab")]
        with self.assertRaisesRegex(ScheduleError, "single string"):
            plan_jobs(jobs)

    def test_non_string_dependency_is_rejected(self):
        with self.assertRaisesRegex(ScheduleError, "non-string"):
            plan_jobs([Job("a"), Job("b", depends_on=(1,))])
        with self.assertRaisesRegex(ScheduleError, "non-string"):
            plan_jobs([Job("a"), Job("b", depends_on=(["a"],))])

    def test_non_iterable_depends_on_is_rejected(self):
        with self.assertRaisesRegex(ScheduleError, "iterable"):
            plan_jobs([Job("a", depends_on=None)])

    def test_non_job_items_are_rejected(self):
        with self.assertRaisesRegex(ScheduleError, "not a Job"):
            plan_jobs([Job("a"), "b"])

    def test_invalid_priority_is_rejected(self):
        for bad in ("high", None, float("nan")):
            with self.subTest(bad=bad):
                with self.assertRaisesRegex(ScheduleError, "priority"):
                    plan_jobs([Job("a", priority=bad)])

    def test_schedule_error_is_a_value_error(self):
        self.assertTrue(issubclass(ScheduleError, ValueError))

    # -- cycles -----------------------------------------------------------

    def test_self_dependency_is_a_cycle(self):
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs([Job("a", depends_on=("a",))])
        self.assertIn("cycle", str(ctx.exception))
        self.assertIn("'a'", str(ctx.exception))

    def test_two_node_cycle(self):
        jobs = [Job("a", depends_on=("b",)), Job("b", depends_on=("a",))]
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs(jobs)
        message = str(ctx.exception)
        self.assertIn("cycle", message)
        self.assertIn("'a'", message)
        self.assertIn("'b'", message)

    def test_cycle_message_lists_the_cycle_path(self):
        jobs = [
            Job("a", depends_on=("b",)),
            Job("b", depends_on=("c",)),
            Job("c", depends_on=("a",)),
        ]
        with self.assertRaisesRegex(
            ScheduleError, r"'a' -> 'b' -> 'c' -> 'a'"
        ):
            plan_jobs(jobs)

    def test_cycle_message_excludes_innocent_and_downstream_jobs(self):
        jobs = [
            Job("ok1"),
            Job("ok2", depends_on=("ok1",)),
            Job("x", depends_on=("y", "ok2")),
            Job("y", depends_on=("x",)),
            Job("after", depends_on=("y",)),
        ]
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs(jobs)
        message = str(ctx.exception)
        self.assertIn("'x'", message)
        self.assertIn("'y'", message)
        for innocent in ("'ok1'", "'ok2'", "'after'"):
            self.assertNotIn(innocent, message)
        self.assertIn("1 more", message)  # "after" is blocked behind the cycle

    def test_cycle_reachable_only_through_a_valid_prefix(self):
        jobs = [
            Job("start"),
            Job("p", depends_on=("start", "q")),
            Job("q", depends_on=("r",)),
            Job("r", depends_on=("p",)),
        ]
        with self.assertRaisesRegex(ScheduleError, "cycle"):
            plan_jobs(jobs)

    def test_cycle_in_second_component_is_found(self):
        jobs = [
            Job("fine"),
            Job("also_fine", depends_on=("fine",)),
            Job("m", depends_on=("n",)),
            Job("n", depends_on=("m",)),
        ]
        with self.assertRaisesRegex(ScheduleError, "'m' -> 'n' -> 'm'"):
            plan_jobs(jobs)

    def test_long_cycle_does_not_hit_recursion_limit(self):
        n = 5000
        jobs = [Job(f"j{i}", depends_on=(f"j{(i + 1) % n}",)) for i in range(n)]
        with self.assertRaisesRegex(ScheduleError, "cycle"):
            plan_jobs(jobs)

    # -- no mutation ------------------------------------------------------

    def test_inputs_are_not_mutated(self):
        deps = ["b", "c"]
        jobs = [
            Job("a", 1, deps, payload={"k": [1, 2]}),
            Job("b", 2),
            Job("c", 3, ["b"]),
        ]
        before = copy.deepcopy(jobs)
        before_list = list(jobs)
        plan_jobs(jobs)
        self.assertEqual(jobs, before)
        self.assertEqual(jobs, before_list)
        self.assertEqual(deps, ["b", "c"])
        for original, current in zip(before_list, jobs):
            self.assertIs(original, current)

    def test_inputs_are_not_mutated_when_planning_fails(self):
        deps = ["b"]
        jobs = [Job("a", depends_on=deps), Job("b", depends_on=["a"])]
        before = copy.deepcopy(jobs)
        with self.assertRaises(ScheduleError):
            plan_jobs(jobs)
        self.assertEqual(jobs, before)
        self.assertEqual(deps, ["b"])

    def test_result_is_a_fresh_list_of_str(self):
        jobs = [Job("a"), Job("b")]
        first = plan_jobs(jobs)
        first.append("junk")
        second = plan_jobs(jobs)
        self.assertEqual(second, ["a", "b"])
        self.assertIsInstance(second, list)
        self.assertTrue(all(type(x) is str for x in second))


if __name__ == "__main__":
    unittest.main()

grok-4.7/01-feature-building/RESPONSE.md

python3 -m unittest -v test_jobqueue.py

Ran 45 tests in 0.010s

OK

grok-4.7/01-feature-building/TASK.md

grok-4.7/01-feature-building/jobqueue.py

import heapq
from collections.abc import Iterable
from dataclasses import dataclass, field
from typing import Any


class ScheduleError(ValueError):
    pass


@dataclass(frozen=True)
class Job:
    id: str
    priority: int = 0
    depends_on: tuple[str, ...] = field(default_factory=tuple)
    payload: Any = None


def plan_jobs(jobs: Iterable[Job]) -> list[str]:
    """Return job IDs in a valid deterministic execution order."""
    try:
        job_list = list(jobs)
    except TypeError as exc:
        raise ScheduleError("jobs must be an iterable of Job") from exc

    if not job_list:
        return []

    # Snapshot every field we need so later scheduling never touches the
    # caller's Job objects or their dependency collections again.
    parsed: list[tuple[str, Any, tuple[Any, ...]]] = []
    positions: dict[str, int] = {}
    for index, job in enumerate(job_list):
        try:
            job_id = job.id
            priority = job.priority
            depends_on = job.depends_on
        except AttributeError as exc:
            raise ScheduleError(f"invalid job at index {index}: {exc}") from exc
        if not isinstance(job_id, str) or job_id == "":
            raise ScheduleError(
                f"job id must be a non-empty string, got {job_id!r}"
            )
        if job_id in positions:
            raise ScheduleError(f"duplicate job id {job_id!r}")
        positions[job_id] = index
        parsed.append((job_id, priority, _snapshot_dependencies(job_id, depends_on)))

    dependencies: dict[str, list[str]] = {}
    dependents: dict[str, list[str]] = {job_id: [] for job_id, _, _ in parsed}
    indegree: dict[str, int] = {}
    for job_id, _priority, raw_deps in parsed:
        # A repeated prerequisite is one edge. Counting it twice would leave
        # the dependent with a leftover indegree and look like a cycle.
        unique: list[str] = []
        seen: set[str] = set()
        for dep in raw_deps:
            if not isinstance(dep, str) or dep not in positions:
                raise ScheduleError(
                    f"job {job_id!r} depends on unknown job {dep!r}"
                )
            if dep in seen:
                continue
            seen.add(dep)
            unique.append(dep)
        dependencies[job_id] = unique
        indegree[job_id] = len(unique)
        for dep in unique:
            dependents[dep].append(job_id)

    # (-priority, input index, id): min-heap pops the highest priority, and
    # for equal priorities the earliest job in the original input.
    ready: list[tuple[Any, int, str]] = []
    for job_id, priority, _raw_deps in parsed:
        if indegree[job_id] == 0:
            _push_ready(ready, priority, positions[job_id], job_id)

    order: list[str] = []
    while ready:
        _neg_priority, _index, job_id = heapq.heappop(ready)
        order.append(job_id)
        for child_id in dependents[job_id]:
            indegree[child_id] -= 1
            if indegree[child_id] == 0:
                child_priority = parsed[positions[child_id]][1]
                _push_ready(ready, child_priority, positions[child_id], child_id)

    if len(order) == len(parsed):
        return order

    scheduled = set(order)
    remaining = [job_id for job_id, _, _ in parsed if job_id not in scheduled]
    cyclic = _cyclic_ids(remaining, dependencies)
    involved = cyclic or remaining
    rendered = ", ".join(repr(job_id) for job_id in involved)
    raise ScheduleError(f"dependency cycle detected involving {rendered}")


def _snapshot_dependencies(job_id: str, depends_on: Any) -> tuple[Any, ...]:
    """Copy dependency ids without mutating or re-reading the caller's collection."""
    if isinstance(depends_on, (str, bytes, bytearray)):
        raise ScheduleError(
            f"job {job_id!r} depends_on must be a collection of job ids, "
            f"not {type(depends_on).__name__}"
        )
    try:
        return tuple(depends_on)
    except TypeError as exc:
        raise ScheduleError(
            f"job {job_id!r} depends_on must be a collection of job ids"
        ) from exc


def _push_ready(
    ready: list[tuple[Any, int, str]], priority: Any, index: int, job_id: str
) -> None:
    try:
        heapq.heappush(ready, (-priority, index, job_id))
    except TypeError as exc:
        raise ScheduleError(
            f"job {job_id!r} priority must be an ordered number, got {priority!r}"
        ) from exc


def _cyclic_ids(
    nodes: list[str], dependencies: dict[str, list[str]]
) -> list[str]:
    """Return cycle participants in input order.

    Strongly connected components of size greater than one, plus self-loops,
    are the jobs on a cycle. A job that only waits on a cycle is omitted.
    """
    node_set = set(nodes)
    forward: dict[str, list[str]] = {}
    reverse: dict[str, list[str]] = {node: [] for node in nodes}
    self_loops: set[str] = set()
    for node in nodes:
        deps: list[str] = []
        for dep in dependencies.get(node, ()):
            if dep not in node_set:
                continue
            if dep == node:
                self_loops.add(node)
            deps.append(dep)
            reverse[dep].append(node)
        forward[node] = deps

    seen: set[str] = set()
    finish: list[str] = []
    for start in nodes:
        if start in seen:
            continue
        stack = [(start, 0)]
        seen.add(start)
        while stack:
            node, cursor = stack[-1]
            children = forward[node]
            if cursor < len(children):
                child = children[cursor]
                stack[-1] = (node, cursor + 1)
                if child not in seen:
                    seen.add(child)
                    stack.append((child, 0))
            else:
                finish.append(node)
                stack.pop()

    seen.clear()
    cyclic: set[str] = set()
    for start in reversed(finish):
        if start in seen:
            continue
        component: list[str] = []
        stack_nodes = [start]
        seen.add(start)
        while stack_nodes:
            node = stack_nodes.pop()
            component.append(node)
            for parent in reverse[node]:
                if parent not in seen:
                    seen.add(parent)
                    stack_nodes.append(parent)
        if len(component) > 1 or any(node in self_loops for node in component):
            cyclic.update(component)

    return [node for node in nodes if node in cyclic]

grok-4.7/01-feature-building/test_jobqueue.py

import random
import unittest

from jobqueue import Job, ScheduleError, plan_jobs


def _ready_set_order(jobs):
    """Independent reading of the scheduling rule for comparison tests.

    Repeatedly select the ready job with the highest priority, breaking ties
    by original input index. Duplicate dependency names count once.
    """
    job_list = list(jobs)
    original_index = {job.id: index for index, job in enumerate(job_list)}
    pending = set(original_index)
    order = []
    satisfied = set()
    while pending:
        ready = []
        for job_id in pending:
            job = job_list[original_index[job_id]]
            seen = set()
            unmet = False
            for dep in job.depends_on:
                if dep in seen:
                    continue
                seen.add(dep)
                if dep not in satisfied:
                    unmet = True
                    break
            if not unmet:
                ready.append(job)
        if not ready:
            raise AssertionError("reference found a cycle in a DAG fixture")
        chosen = min(
            ready,
            key=lambda job: (-job.priority, original_index[job.id]),
        )
        order.append(chosen.id)
        pending.remove(chosen.id)
        satisfied.add(chosen.id)
    return order


class PlanJobsTests(unittest.TestCase):
    maxDiff = None

    def test_dependencies_override_priority(self):
        jobs = [
            Job("build", priority=100, depends_on=("fetch",)),
            Job("fetch", priority=1),
        ]
        self.assertEqual(plan_jobs(jobs), ["fetch", "build"])

    def test_ready_jobs_use_priority_then_input_order(self):
        jobs = [Job("a", 2), Job("b", 5), Job("c", 2)]
        self.assertEqual(plan_jobs(jobs), ["b", "a", "c"])

    def test_unknown_dependency_is_rejected(self):
        with self.assertRaises(ScheduleError):
            plan_jobs([Job("build", depends_on=("missing",))])

    def test_empty_input_returns_empty_list(self):
        self.assertEqual(plan_jobs([]), [])
        self.assertEqual(plan_jobs(()), [])

        def generate():
            if False:
                yield Job("unused")

        self.assertEqual(plan_jobs(generate()), [])

    def test_single_job(self):
        payload = {"path": "src"}
        job = Job("only", payload=payload)
        self.assertEqual(plan_jobs([job]), ["only"])
        self.assertIs(job.payload, payload)

    def test_blank_but_non_empty_id_is_allowed(self):
        self.assertEqual(plan_jobs([Job(" ")]), [" "])

    def test_empty_id_is_rejected(self):
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs([Job("")])
        self.assertIn("non-empty", str(ctx.exception))

    def test_non_string_ids_are_rejected(self):
        for bad_id in (None, 0, b"a", ("a",)):
            with self.subTest(bad_id=bad_id):
                with self.assertRaises(ScheduleError) as ctx:
                    plan_jobs([Job(bad_id)])
                message = str(ctx.exception)
                self.assertIn("non-empty", message)
                self.assertIn(repr(bad_id), message)

    def test_duplicate_id_is_rejected_before_scheduling(self):
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs(
                [
                    Job("a", depends_on=("a",)),
                    Job("b"),
                    Job("a", priority=9),
                ]
            )
        message = str(ctx.exception)
        self.assertIn("duplicate", message)
        self.assertIn("'a'", message)
        self.assertNotIn("cycle", message)

    def test_unknown_dependency_names_both_jobs(self):
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs(
                [
                    Job("build", depends_on=("fetch", "missing")),
                    Job("fetch"),
                    Job("later", depends_on=("also-missing",)),
                ]
            )
        message = str(ctx.exception)
        self.assertIn("unknown", message)
        self.assertIn("'build'", message)
        self.assertIn("'missing'", message)
        self.assertNotIn("also-missing", message)

    def test_dependency_match_is_exact_not_stringified(self):
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs([Job("1"), Job("next", depends_on=(1,))])
        message = str(ctx.exception)
        self.assertIn("unknown", message)
        self.assertIn("'next'", message)
        self.assertIn("1", message)

    def test_string_depends_on_is_rejected(self):
        # Iterating the string "ab" would look like dependencies "a" and "b".
        for value in ("ab", "a", b"ab"):
            with self.subTest(value=value):
                with self.assertRaises(ScheduleError) as ctx:
                    plan_jobs([Job("ab", depends_on=value), Job("a"), Job("b")])
                message = str(ctx.exception)
                self.assertIn("depends_on", message)
                self.assertNotIn("cycle", message)

    def test_non_iterable_jobs_are_rejected(self):
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs(None)
        self.assertIn("iterable", str(ctx.exception))
        with self.assertRaises(ScheduleError):
            plan_jobs(Job("a"))

    def test_non_iterable_depends_on_is_rejected(self):
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs([Job("a", depends_on=123)])
        self.assertIn("depends_on", str(ctx.exception))

    def test_incomparable_priorities_are_rejected(self):
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs([Job("a", priority=1), Job("b", priority="high")])
        message = str(ctx.exception)
        self.assertIn("priority", message)
        self.assertIn("'b'", message)

    def test_forward_dependency_is_honored(self):
        jobs = [
            Job("child", priority=50, depends_on=("parent",)),
            Job("parent", priority=0),
        ]
        self.assertEqual(plan_jobs(jobs), ["parent", "child"])

    def test_chain_keeps_high_priority_job_behind_its_dependencies(self):
        jobs = [
            Job("a", priority=1, depends_on=("b",)),
            Job("b", priority=10, depends_on=("c",)),
            Job("c", priority=100),
        ]
        self.assertEqual(plan_jobs(jobs), ["c", "b", "a"])

    def test_negative_priorities_rank_below_zero(self):
        jobs = [Job("a", priority=-5), Job("b", priority=-1), Job("c", priority=0)]
        self.assertEqual(plan_jobs(jobs), ["c", "b", "a"])

    def test_equal_priorities_keep_input_order(self):
        jobs = [Job("a", priority=5), Job("b", priority=5), Job("c", priority=5)]
        self.assertEqual(plan_jobs(jobs), ["a", "b", "c"])

    def test_equal_priority_uses_input_order_not_ready_time(self):
        # b is ready immediately. a becomes ready only after c, but a appears
        # earlier in the input and has the same priority, so a goes before b.
        jobs = [
            Job("a", priority=1, depends_on=("c",)),
            Job("b", priority=1),
            Job("c", priority=5),
        ]
        self.assertEqual(plan_jobs(jobs), ["c", "a", "b"])

    def test_newly_ready_higher_priority_jumps_ahead(self):
        jobs = [
            Job("slow", priority=1),
            Job("gate", priority=2),
            Job("fast", priority=50, depends_on=("gate",)),
        ]
        self.assertEqual(plan_jobs(jobs), ["gate", "fast", "slow"])

    def test_lower_priority_dependency_runs_before_a_waiting_high_priority_job(self):
        jobs = [
            Job("low", priority=1),
            Job("blocker", priority=0),
            Job("high", priority=100, depends_on=("blocker",)),
        ]
        self.assertEqual(plan_jobs(jobs), ["low", "blocker", "high"])

    def test_same_moment_ready_jobs_follow_input_order_not_dependency_list_order(self):
        deps = ["left", "right"]
        jobs = [
            Job("top", priority=1, depends_on=deps),
            Job("right", priority=5),
            Job("left", priority=5),
        ]
        self.assertEqual(plan_jobs(jobs), ["right", "left", "top"])
        self.assertEqual(deps, ["left", "right"])

    def test_fan_in_waits_for_every_dependency(self):
        jobs = [
            Job("final", priority=100, depends_on=("a", "b", "c")),
            Job("b", priority=2),
            Job("c", priority=3),
            Job("a", priority=1),
        ]
        self.assertEqual(plan_jobs(jobs), ["c", "b", "a", "final"])

    def test_fan_out_orders_siblings_by_priority_then_input(self):
        jobs = [
            Job("a", depends_on=("root",)),
            Job("b", priority=10, depends_on=("root",)),
            Job("c", priority=10, depends_on=("root",)),
            Job("root"),
        ]
        self.assertEqual(plan_jobs(jobs), ["root", "b", "c", "a"])

    def test_diamond_dependency(self):
        jobs = [
            Job("d", priority=1, depends_on=("b", "a")),
            Job("a", priority=3, depends_on=("c",)),
            Job("b", priority=3, depends_on=("c",)),
            Job("c", priority=0),
        ]
        self.assertEqual(plan_jobs(jobs), ["c", "a", "b", "d"])

    def test_unlocked_jobs_do_not_pass_an_earlier_equal_priority_job(self):
        jobs = [
            Job("a"),
            Job("b", priority=-1),
            Job("c", depends_on=("a",)),
            Job("d", depends_on=("a",)),
        ]
        self.assertEqual(plan_jobs(jobs), ["a", "c", "d", "b"])

    def test_duplicate_dependency_edges_are_not_a_cycle(self):
        jobs = [
            Job("a", priority=100, depends_on=("b", "b")),
            Job("b", priority=0),
        ]
        self.assertEqual(plan_jobs(jobs), ["b", "a"])

    def test_payload_does_not_affect_order(self):
        jobs = [
            Job("a", priority=1, payload={"heavy": True}),
            Job("b", priority=2, payload=["x"]),
        ]
        self.assertEqual(plan_jobs(jobs), ["b", "a"])

    def test_accepts_one_shot_iterable(self):
        def generate():
            yield Job("b", priority=1, depends_on=("a",))
            yield Job("a", priority=0)

        self.assertEqual(plan_jobs(generate()), ["a", "b"])

    def test_repeated_calls_are_deterministic(self):
        jobs = [
            Job("a", priority=1, depends_on=("c",)),
            Job("b", priority=1),
            Job("c", priority=4),
            Job("d", priority=4, depends_on=("b",)),
        ]
        first = plan_jobs(jobs)
        self.assertEqual(first, ["c", "a", "b", "d"])
        self.assertEqual(plan_jobs(tuple(jobs)), first)
        self.assertEqual(plan_jobs(iter(jobs)), first)
        first.append("extra")
        self.assertEqual(plan_jobs(jobs), ["c", "a", "b", "d"])

    def test_self_cycle_names_the_job(self):
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs([Job("solo", depends_on=("solo",))])
        self.assertEqual(
            str(ctx.exception),
            "dependency cycle detected involving 'solo'",
        )

    def test_self_cycle_with_an_outside_dependency(self):
        jobs = [
            Job("a", depends_on=("a", "c")),
            Job("c", priority=2),
            Job("b", priority=1),
        ]
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs(jobs)
        self.assertEqual(
            str(ctx.exception),
            "dependency cycle detected involving 'a'",
        )

    def test_two_node_cycle_names_both_jobs(self):
        jobs = [
            Job("a", depends_on=("b",)),
            Job("b", depends_on=("a",)),
        ]
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs(jobs)
        self.assertEqual(
            str(ctx.exception),
            "dependency cycle detected involving 'a', 'b'",
        )

    def test_three_node_cycle_names_every_participant(self):
        jobs = [
            Job("a", depends_on=("b",)),
            Job("b", depends_on=("c",)),
            Job("c", depends_on=("a",)),
        ]
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs(jobs)
        self.assertEqual(
            str(ctx.exception),
            "dependency cycle detected involving 'a', 'b', 'c'",
        )

    def test_disjoint_cycles_name_every_participant(self):
        jobs = [
            Job("a", depends_on=("b",)),
            Job("b", depends_on=("a",)),
            Job("c", depends_on=("d",)),
            Job("d", depends_on=("c",)),
        ]
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs(jobs)
        self.assertEqual(
            str(ctx.exception),
            "dependency cycle detected involving 'a', 'b', 'c', 'd'",
        )

    def test_cycle_diagnostic_omits_runnable_and_merely_blocked_jobs(self):
        jobs = [
            Job("outside", priority=100),
            Job("a", depends_on=("b", "base")),
            Job("b", depends_on=("a",)),
            Job("blocked", depends_on=("a",)),
            Job("base", priority=1),
        ]
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs(jobs)
        self.assertEqual(
            str(ctx.exception),
            "dependency cycle detected involving 'a', 'b'",
        )

    def test_self_loop_plus_mutual_dependency(self):
        jobs = [
            Job("a", depends_on=("a", "b")),
            Job("b", depends_on=("a",)),
        ]
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs(jobs)
        self.assertEqual(
            str(ctx.exception),
            "dependency cycle detected involving 'a', 'b'",
        )

    def test_unknown_dependency_is_reported_before_cycle_check(self):
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs([Job("a", depends_on=("a", "missing"))])
        message = str(ctx.exception)
        self.assertIn("unknown", message)
        self.assertIn("'missing'", message)
        self.assertNotIn("cycle", message)

    def test_schedule_error_is_a_value_error(self):
        with self.assertRaises(ValueError):
            plan_jobs([Job("a", depends_on=("missing",))])

    def test_does_not_mutate_inputs_on_success_or_failure(self):
        success_deps = ["b", "b"]
        success_payload = {"n": 1}
        success_jobs = [
            Job("a", priority=1, depends_on=success_deps, payload=success_payload),
            Job("b", priority=2, payload=["keep"]),
        ]
        failure_deps = ["a"]
        failure_jobs = [
            Job("a", depends_on=failure_deps),
            Job("free", priority=3),
        ]
        success_before = _snapshot(success_jobs)
        failure_before = _snapshot(failure_jobs)
        success_order_before = [job.id for job in success_jobs]
        failure_order_before = [job.id for job in failure_jobs]

        self.assertEqual(plan_jobs(success_jobs), ["b", "a"])
        with self.assertRaises(ScheduleError):
            plan_jobs(failure_jobs)

        self.assertEqual(_snapshot(success_jobs), success_before)
        self.assertEqual(_snapshot(failure_jobs), failure_before)
        self.assertEqual([job.id for job in success_jobs], success_order_before)
        self.assertEqual([job.id for job in failure_jobs], failure_order_before)
        self.assertIs(success_jobs[0].depends_on, success_deps)
        self.assertIs(failure_jobs[0].depends_on, failure_deps)
        self.assertIs(success_jobs[0].payload, success_payload)
        self.assertEqual(success_deps, ["b", "b"])
        self.assertEqual(failure_deps, ["a"])
        self.assertEqual(success_payload, {"n": 1})

    def test_long_chain_respects_dependencies_over_priority(self):
        count = 1200
        jobs = [
            Job(str(i), priority=i, depends_on=((str(i - 1),) if i else ()))
            for i in range(count)
        ]
        self.assertEqual(plan_jobs(jobs), [str(i) for i in range(count)])

    def test_long_cycle_reports_members_without_recursion_limit(self):
        count = 1200
        jobs = [
            Job(str(i), depends_on=(str((i + 1) % count),))
            for i in range(count)
        ]
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs(jobs)
        message = str(ctx.exception)
        self.assertIn("'0'", message)
        self.assertIn(f"'{count - 1}'", message)
        self.assertIn("cycle", message)

    def test_random_dags_match_ready_set_rule(self):
        rng = random.Random(0)
        for trial in range(40):
            count = rng.randint(1, 30)
            ranked_ids = [f"j{index:02d}" for index in range(count)]
            input_ids = ranked_ids[:]
            rng.shuffle(input_ids)
            rank = {job_id: index for index, job_id in enumerate(ranked_ids)}
            jobs = []
            for job_id in input_ids:
                candidates = [
                    other for other in ranked_ids if rank[other] < rank[job_id]
                ]
                dep_count = rng.randint(0, min(3, len(candidates)))
                deps = tuple(rng.sample(candidates, dep_count)) if dep_count else ()
                if deps and rng.random() < 0.4:
                    deps = deps + (deps[0],)
                jobs.append(
                    Job(job_id, priority=rng.randint(-3, 6), depends_on=deps)
                )
            with self.subTest(trial=trial, count=count):
                order = plan_jobs(jobs)
                self.assertEqual(order, _ready_set_order(jobs))
                self._assert_respects_dependencies(jobs, order)

    def test_random_cycles_name_every_participant(self):
        rng = random.Random(1)
        for trial in range(15):
            cycle_len = rng.randint(1, 8)
            other_count = rng.randint(0, 5)
            cycle_ids = [f"cycle{index:02d}" for index in range(cycle_len)]
            other_ids = [f"other{index:02d}" for index in range(other_count)]
            cycled = [
                Job(
                    cycle_ids[index],
                    priority=rng.randint(0, 5),
                    depends_on=(cycle_ids[(index + 1) % cycle_len],),
                )
                for index in range(cycle_len)
            ]
            others = [
                Job(other_id, priority=rng.randint(0, 5)) for other_id in other_ids
            ]
            blocked = Job("blocked", priority=9, depends_on=(cycle_ids[0],))
            jobs = [*others, blocked, *cycled]
            rng.shuffle(jobs)
            with self.subTest(trial=trial, cycle_len=cycle_len):
                with self.assertRaises(ScheduleError) as ctx:
                    plan_jobs(jobs)
                message = str(ctx.exception)
                for job_id in cycle_ids:
                    self.assertIn(repr(job_id), message)
                for job_id in other_ids:
                    self.assertNotIn(repr(job_id), message)
                self.assertNotIn(repr("blocked"), message)

    def _assert_respects_dependencies(self, jobs, order):
        self.assertEqual(len(order), len({job.id for job in jobs}))
        self.assertEqual(set(order), {job.id for job in jobs})
        position = {job_id: index for index, job_id in enumerate(order)}
        for job in jobs:
            seen = set()
            for dep in job.depends_on:
                if dep in seen:
                    continue
                seen.add(dep)
                self.assertLess(position[dep], position[job.id])


def _snapshot(jobs):
    return [
        (job.id, job.priority, tuple(job.depends_on), job.payload) for job in jobs
    ]


if __name__ == "__main__":
    unittest.main()

opus-5.5/01-feature-building/RESPONSE.md

python3 -m unittest -v test_jobqueue

Ran 34 tests in 0.038s

OK

opus-5.5/01-feature-building/TASK.md

opus-5.5/01-feature-building/jobqueue.py

import heapq
from dataclasses import dataclass, field
from typing import Any, Iterable


class ScheduleError(ValueError):
    pass


@dataclass(frozen=True)
class Job:
    id: str
    priority: int = 0
    depends_on: tuple[str, ...] = field(default_factory=tuple)
    payload: Any = None


def _validate(jobs: list[Job]) -> dict[str, int]:
    """Validate jobs and return a mapping of job ID -> input index."""
    index: dict[str, int] = {}
    for position, job in enumerate(jobs):
        if not isinstance(job, Job):
            raise ScheduleError(
                f"item at position {position} is not a Job: {job!r}"
            )
        if not isinstance(job.id, str) or not job.id:
            raise ScheduleError(
                f"job at position {position} has an invalid id {job.id!r}; "
                "ids must be non-empty strings"
            )
        if job.id in index:
            raise ScheduleError(
                f"duplicate job id {job.id!r} at positions "
                f"{index[job.id]} and {position}"
            )
        if isinstance(job.priority, bool) or not isinstance(job.priority, int):
            raise ScheduleError(
                f"job {job.id!r} has a non-integer priority {job.priority!r}"
            )
        if isinstance(job.depends_on, (str, bytes)):
            raise ScheduleError(
                f"job {job.id!r} depends_on must be a sequence of job ids, "
                f"not a single string {job.depends_on!r}"
            )
        try:
            deps = tuple(job.depends_on)
        except TypeError:
            raise ScheduleError(
                f"job {job.id!r} depends_on is not iterable: {job.depends_on!r}"
            ) from None
        for dep in deps:
            if not isinstance(dep, str) or not dep:
                raise ScheduleError(
                    f"job {job.id!r} has an invalid dependency {dep!r}; "
                    "dependencies must be non-empty strings"
                )
        index[job.id] = position

    for job in jobs:
        missing = sorted({dep for dep in job.depends_on if dep not in index})
        if missing:
            raise ScheduleError(
                f"job {job.id!r} depends on unknown job(s): "
                + ", ".join(repr(m) for m in missing)
            )
    return index


def _find_cycle(blocked: list[str], deps: dict[str, set[str]]) -> list[str]:
    """Return one dependency cycle among ``blocked`` as [a, b, ..., a].

    Every blocked job has at least one blocked dependency, so walking
    dependency edges from any blocked job must eventually revisit a job.
    """
    blocked_set = set(blocked)
    path: list[str] = []
    seen_at: dict[str, int] = {}
    node = blocked[0]
    while node not in seen_at:
        seen_at[node] = len(path)
        path.append(node)
        # Deterministic choice: smallest blocked dependency ID.
        node = min(d for d in deps[node] if d in blocked_set)
    cycle = path[seen_at[node]:]
    # Present as "X depends on Y depends on ... depends on X".
    return cycle + [node]


def plan_jobs(jobs: Iterable[Job]) -> list[str]:
    """Return job IDs in a valid deterministic execution order.

    Every job appears exactly once and after all of its dependencies. Among
    jobs that are ready at the same time, higher ``priority`` runs first and
    ties are broken by original input order.

    Raises ``ScheduleError`` for non-string/empty/duplicate IDs, unknown
    dependencies, or dependency cycles. The input jobs are never mutated.
    """
    job_list = list(jobs)
    if not job_list:
        return []

    index = _validate(job_list)

    # deps[x]: distinct jobs x waits on; dependents[y]: jobs waiting on y.
    deps: dict[str, set[str]] = {job.id: set(job.depends_on) for job in job_list}
    dependents: dict[str, list[str]] = {job.id: [] for job in job_list}
    for job in job_list:
        for dep in deps[job.id]:
            dependents[dep].append(job.id)
    remaining = {job_id: len(d) for job_id, d in deps.items()}

    # Heap key: (-priority, input position) => higher priority, then earlier.
    ready = [
        (-job.priority, index[job.id], job.id)
        for job in job_list
        if remaining[job.id] == 0
    ]
    heapq.heapify(ready)

    by_id = {job.id: job for job in job_list}
    order: list[str] = []
    while ready:
        _, _, job_id = heapq.heappop(ready)
        order.append(job_id)
        for child in dependents[job_id]:
            remaining[child] -= 1
            if remaining[child] == 0:
                child_job = by_id[child]
                heapq.heappush(
                    ready, (-child_job.priority, index[child], child)
                )

    if len(order) != len(job_list):
        blocked = [job.id for job in job_list if remaining[job.id] > 0]
        cycle = _find_cycle(blocked, deps)
        raise ScheduleError(
            "dependency cycle detected ('->' means 'depends on'): "
            + " -> ".join(cycle)
            + "; blocked jobs: "
            + ", ".join(blocked)
        )
    return order

opus-5.5/01-feature-building/test_jobqueue.py

import copy
import random
import unittest

from jobqueue import Job, ScheduleError, plan_jobs


def assert_valid_order(test, jobs, order):
    """Every job exactly once, and each after all of its dependencies."""
    ids = [job.id for job in jobs]
    test.assertCountEqual(order, ids)
    test.assertEqual(len(order), len(set(order)))
    position = {job_id: i for i, job_id in enumerate(order)}
    for job in jobs:
        for dep in job.depends_on:
            test.assertLess(
                position[dep], position[job.id], f"{dep} must precede {job.id}"
            )


class PlanJobsTests(unittest.TestCase):
    # --- Original cases -------------------------------------------------
    def test_dependencies_override_priority(self):
        jobs = [
            Job("build", priority=100, depends_on=("fetch",)),
            Job("fetch", priority=1),
        ]
        self.assertEqual(plan_jobs(jobs), ["fetch", "build"])

    def test_ready_jobs_use_priority_then_input_order(self):
        jobs = [Job("a", 2), Job("b", 5), Job("c", 2)]
        self.assertEqual(plan_jobs(jobs), ["b", "a", "c"])

    def test_unknown_dependency_is_rejected(self):
        with self.assertRaises(ScheduleError):
            plan_jobs([Job("build", depends_on=("missing",))])

    # --- Basic shapes ---------------------------------------------------
    def test_empty_input_returns_empty_list(self):
        self.assertEqual(plan_jobs([]), [])
        self.assertEqual(plan_jobs(iter(())), [])

    def test_single_job(self):
        self.assertEqual(plan_jobs([Job("only")]), ["only"])

    def test_accepts_generator_input(self):
        jobs = (Job(name, p) for name, p in [("x", 1), ("y", 3), ("z", 2)])
        self.assertEqual(plan_jobs(jobs), ["y", "z", "x"])

    def test_returns_new_list_each_call(self):
        jobs = [Job("a"), Job("b")]
        first = plan_jobs(jobs)
        first.append("junk")
        self.assertEqual(plan_jobs(jobs), ["a", "b"])

    # --- Ordering -------------------------------------------------------
    def test_all_equal_priority_preserves_input_order(self):
        jobs = [Job(name) for name in "qwerty"]
        self.assertEqual(plan_jobs(jobs), list("qwerty"))

    def test_negative_and_zero_priorities(self):
        jobs = [Job("low", -5), Job("zero", 0), Job("lower", -10), Job("high", 3)]
        self.assertEqual(plan_jobs(jobs), ["high", "zero", "low", "lower"])

    def test_newly_ready_high_priority_job_jumps_queue(self):
        # After "a" finishes, "urgent" becomes ready and outranks "b"/"c",
        # which were ready from the start.
        jobs = [
            Job("a", 10),
            Job("b", 5),
            Job("c", 5),
            Job("urgent", 50, depends_on=("a",)),
        ]
        self.assertEqual(plan_jobs(jobs), ["a", "urgent", "b", "c"])

    def test_newly_ready_job_ties_broken_by_input_order_not_readiness(self):
        # "late" is listed first, so once it becomes ready it beats "early"
        # at equal priority even though "early" has been waiting longer.
        jobs = [
            Job("late", 1, depends_on=("root",)),
            Job("root", 9),
            Job("early", 1),
        ]
        self.assertEqual(plan_jobs(jobs), ["root", "late", "early"])

    def test_low_priority_dependency_is_not_starved(self):
        jobs = [
            Job("deploy", 100, depends_on=("test",)),
            Job("lint", 50),
            Job("test", -1),
        ]
        self.assertEqual(plan_jobs(jobs), ["lint", "test", "deploy"])

    def test_diamond(self):
        jobs = [
            Job("d", 0, depends_on=("b", "c")),
            Job("b", 1, depends_on=("a",)),
            Job("c", 2, depends_on=("a",)),
            Job("a", 0),
        ]
        self.assertEqual(plan_jobs(jobs), ["a", "c", "b", "d"])

    def test_long_chain_listed_in_reverse(self):
        n = 50
        jobs = [
            Job(f"j{i}", depends_on=(f"j{i - 1}",) if i else ())
            for i in reversed(range(n))
        ]
        self.assertEqual(plan_jobs(jobs), [f"j{i}" for i in range(n)])

    def test_disconnected_components_interleave_by_priority(self):
        jobs = [
            Job("a1", 1),
            Job("a2", 1, depends_on=("a1",)),
            Job("b1", 2),
            Job("b2", 0, depends_on=("b1",)),
        ]
        self.assertEqual(plan_jobs(jobs), ["b1", "a1", "a2", "b2"])

    def test_duplicate_dependency_entries_are_tolerated(self):
        jobs = [Job("a"), Job("b", depends_on=("a", "a"))]
        self.assertEqual(plan_jobs(jobs), ["a", "b"])

    def test_list_depends_on_is_accepted(self):
        jobs = [Job("b", depends_on=["a"]), Job("a")]
        self.assertEqual(plan_jobs(jobs), ["a", "b"])

    def test_deterministic_across_calls(self):
        jobs = [Job(str(i), i % 3) for i in range(20)]
        self.assertEqual(plan_jobs(jobs), plan_jobs(jobs))

    def test_random_dags_produce_valid_orders(self):
        rng = random.Random(1234)
        for _ in range(200):
            n = rng.randint(1, 15)
            jobs = []
            for i in range(n):
                deps = tuple(
                    f"n{j}" for j in range(i) if rng.random() < 0.3
                )
                jobs.append(Job(f"n{i}", rng.randint(-2, 2), deps))
            rng.shuffle(jobs)
            order = plan_jobs(jobs)
            assert_valid_order(self, jobs, order)
            self._assert_greedy_priority(jobs, order)

    def _assert_greedy_priority(self, jobs, order):
        """Each emitted job is the best-ranked job ready at that moment."""
        position = {job.id: i for i, job in enumerate(jobs)}
        done = set()
        for job_id in order:
            ready = [
                j for j in jobs
                if j.id not in done and all(d in done for d in j.depends_on)
            ]
            best = min(ready, key=lambda j: (-j.priority, position[j.id]))
            self.assertEqual(job_id, best.id)
            done.add(job_id)

    # --- Validation -----------------------------------------------------
    def test_unknown_dependency_message_names_job_and_dependency(self):
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs([Job("build", depends_on=("missing",))])
        self.assertIn("build", str(ctx.exception))
        self.assertIn("missing", str(ctx.exception))

    def test_duplicate_ids_rejected(self):
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs([Job("a"), Job("b"), Job("a")])
        self.assertIn("'a'", str(ctx.exception))
        self.assertIn("duplicate", str(ctx.exception))

    def test_empty_id_rejected(self):
        with self.assertRaises(ScheduleError):
            plan_jobs([Job("")])

    def test_non_string_ids_rejected(self):
        for bad in (None, 1, b"bytes", ("a",)):
            with self.subTest(bad=bad):
                with self.assertRaises(ScheduleError):
                    plan_jobs([Job(bad)])

    def test_non_job_items_rejected(self):
        with self.assertRaises(ScheduleError):
            plan_jobs(["a"])

    def test_string_depends_on_rejected(self):
        # A bare string would otherwise be iterated character by character.
        with self.assertRaises(ScheduleError):
            plan_jobs([Job("a"), Job("ab", depends_on="a")])

    def test_invalid_dependency_entries_rejected(self):
        for bad in (None, 3, ""):
            with self.subTest(bad=bad):
                with self.assertRaises(ScheduleError):
                    plan_jobs([Job("a"), Job("b", depends_on=(bad,))])

    def test_non_integer_priority_rejected(self):
        for bad in ("high", None, 1.5, True):
            with self.subTest(bad=bad):
                with self.assertRaises(ScheduleError):
                    plan_jobs([Job("a", priority=bad)])

    def test_schedule_error_is_value_error(self):
        with self.assertRaises(ValueError):
            plan_jobs([Job("a", depends_on=("zzz",))])

    # --- Cycles ---------------------------------------------------------
    def test_self_dependency_is_a_cycle(self):
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs([Job("loop", depends_on=("loop",))])
        self.assertIn("cycle", str(ctx.exception))
        self.assertIn("loop", str(ctx.exception))

    def test_two_node_cycle_names_both_ids(self):
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs([Job("a", depends_on=("b",)), Job("b", depends_on=("a",))])
        message = str(ctx.exception)
        self.assertIn("cycle", message)
        self.assertIn("a -> b -> a", message)

    def test_cycle_diagnostic_excludes_unrelated_jobs(self):
        jobs = [
            Job("ok"),
            Job("x", depends_on=("ok", "z")),
            Job("y", depends_on=("x",)),
            Job("z", depends_on=("y",)),
            Job("after", depends_on=("x",)),
        ]
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs(jobs)
        message = str(ctx.exception)
        cycle_part = message.split(";")[0]
        for job_id in ("x", "y", "z"):
            self.assertIn(job_id, cycle_part)
        self.assertNotIn("ok", message)
        self.assertNotIn("after", cycle_part)
        # Jobs downstream of the cycle are reported as blocked.
        self.assertIn("after", message.split(";")[1])

    def test_cycle_detected_even_when_other_jobs_are_schedulable(self):
        jobs = [Job(str(i)) for i in range(10)] + [
            Job("p", depends_on=("q",)),
            Job("q", depends_on=("p",)),
        ]
        with self.assertRaises(ScheduleError):
            plan_jobs(jobs)

    # --- Immutability ---------------------------------------------------
    def test_inputs_are_not_mutated(self):
        deps_list = ["a"]
        jobs = [
            Job("b", 1, depends_on=deps_list, payload={"k": [1]}),
            Job("a", 0),
            Job("c", 5, depends_on=("a", "b")),
        ]
        input_list = list(jobs)
        snapshot = copy.deepcopy(jobs)
        plan_jobs(jobs)
        self.assertEqual(jobs, input_list)
        self.assertEqual(jobs, snapshot)
        self.assertEqual(deps_list, ["a"])
        self.assertIs(jobs[0].depends_on, deps_list)

    def test_inputs_not_mutated_on_error(self):
        deps_list = ["b"]
        jobs = [Job("a", depends_on=deps_list), Job("b", depends_on=("a",))]
        snapshot = copy.deepcopy(jobs)
        with self.assertRaises(ScheduleError):
            plan_jobs(jobs)
        self.assertEqual(jobs, snapshot)
        self.assertEqual(deps_list, ["b"])


if __name__ == "__main__":
    unittest.main()

gpt-6-astra/01-feature-building/RESPONSE.md

python3 -B -m unittest -v test_jobqueue

Ran 25 tests in 0.005s

OK

gpt-6-astra/01-feature-building/TASK.md

gpt-6-astra/01-feature-building/jobqueue.py

from dataclasses import dataclass, field
from heapq import heapify, heappop, heappush
from typing import Any, Iterable


class ScheduleError(ValueError):
    pass


@dataclass(frozen=True)
class Job:
    id: str
    priority: int = 0
    depends_on: tuple[str, ...] = field(default_factory=tuple)
    payload: Any = None


def plan_jobs(jobs: Iterable[Job]) -> list[str]:
    """Order jobs by dependencies, then ready priority and original input order.

    Repeated dependencies count as one prerequisite. Cycle diagnostics list
    blocked jobs, which can include downstream dependents of the cycle.
    """
    jobs = list(jobs)
    indices: dict[str, int] = {}
    for index, job in enumerate(jobs):
        if not isinstance(job.id, str) or not job.id:
            raise ScheduleError(
                f"Job at input index {index} must have a non-empty string ID; "
                f"got {job.id!r}"
            )
        if job.id in indices:
            raise ScheduleError(f"Duplicate job ID: {job.id!r}")
        indices[job.id] = index

    remaining = [0] * len(jobs)
    dependents: list[list[int]] = [[] for _ in jobs]
    for index, job in enumerate(jobs):
        seen: set[str] = set()
        for dependency in job.depends_on:
            if not isinstance(dependency, str) or not dependency:
                raise ScheduleError(
                    f"Job {job.id!r} has invalid dependency {dependency!r}; "
                    "expected a non-empty string job ID"
                )
            if dependency not in indices:
                raise ScheduleError(
                    f"Job {job.id!r} has unknown dependency {dependency!r}"
                )
            if dependency not in seen:
                seen.add(dependency)
                remaining[index] += 1
                dependents[indices[dependency]].append(index)

    ready = [
        (-job.priority, index)
        for index, job in enumerate(jobs)
        if remaining[index] == 0
    ]
    heapify(ready)
    ordered: list[str] = []
    while ready:
        _, index = heappop(ready)
        ordered.append(jobs[index].id)
        for dependent in dependents[index]:
            remaining[dependent] -= 1
            if remaining[dependent] == 0:
                heappush(ready, (-jobs[dependent].priority, dependent))

    if len(ordered) != len(jobs):
        blocked = ", ".join(
            repr(jobs[index].id)
            for index, count in enumerate(remaining)
            if count > 0
        )
        raise ScheduleError(f"Dependency cycle detected; blocked job IDs: {blocked}")
    return ordered

gpt-6-astra/01-feature-building/test_jobqueue.py

from copy import deepcopy
from itertools import combinations, permutations
import unittest

from jobqueue import Job, ScheduleError, plan_jobs


class PlanJobsTests(unittest.TestCase):
    def test_empty_input(self):
        self.assertEqual(plan_jobs([]), [])
        self.assertEqual(plan_jobs(iter(())), [])

    def test_single_job(self):
        self.assertEqual(plan_jobs([Job("only")]), ["only"])

    def test_dependencies_override_priority(self):
        jobs = [
            Job("build", priority=100, depends_on=("fetch",)),
            Job("fetch", priority=1),
        ]
        self.assertEqual(plan_jobs(jobs), ["fetch", "build"])

    def test_ready_jobs_use_priority_then_input_order(self):
        jobs = [Job("a", 2), Job("b", 5), Job("c", 2)]
        self.assertEqual(plan_jobs(jobs), ["b", "a", "c"])

    def test_unknown_dependency_is_rejected(self):
        with self.assertRaisesRegex(ScheduleError, "build.*missing"):
            plan_jobs([Job("build", depends_on=("missing",))])

    def test_invalid_job_ids_are_rejected(self):
        for invalid_id in ("", None, 42, True, [], {}):
            with self.subTest(job_id=invalid_id):
                with self.assertRaisesRegex(ScheduleError, "non-empty string ID"):
                    plan_jobs([Job(invalid_id)])

    def test_duplicate_job_ids_are_rejected(self):
        with self.assertRaisesRegex(ScheduleError, "Duplicate.*same"):
            plan_jobs([Job("same", 1), Job("same", 100)])

    def test_invalid_dependencies_are_rejected(self):
        for dependency in ("", None, 42, True, [], {}):
            with self.subTest(dependency=dependency):
                with self.assertRaisesRegex(ScheduleError, "build.*invalid dependency"):
                    plan_jobs([Job("build", depends_on=(dependency,))])

    def test_all_components_are_validated(self):
        with self.assertRaisesRegex(ScheduleError, "broken.*missing"):
            plan_jobs([Job("ready", 100), Job("broken", depends_on=("missing",))])

    def test_nonempty_ids_are_not_normalized(self):
        jobs = [Job(" a "), Job("a"), Job(" "), Job("猫")]
        self.assertEqual(plan_jobs(jobs), [" a ", "a", " ", "猫"])

    def test_equal_priorities_use_input_order_not_id_order(self):
        jobs = [Job("z"), Job("m"), Job("a")]
        self.assertEqual(plan_jobs(jobs), ["z", "m", "a"])

    def test_negative_priorities(self):
        jobs = [Job("low", -10), Job("medium", -1), Job("high", 0)]
        self.assertEqual(plan_jobs(jobs), ["high", "medium", "low"])

    def test_newly_ready_job_preempts_lower_priority_job(self):
        jobs = [Job("waiting", 1), Job("root", 2), Job("child", 100, ("root",))]
        self.assertEqual(plan_jobs(jobs), ["root", "child", "waiting"])

    def test_newly_ready_tie_uses_original_order_not_ready_time(self):
        jobs = [Job("child", 1, ("root",)), Job("waiting", 1), Job("root", 2)]
        self.assertEqual(plan_jobs(jobs), ["root", "child", "waiting"])

    def test_simultaneously_ready_jobs_use_input_order(self):
        jobs = [Job("z", 1, ("root",)), Job("a", 1, ("root",)), Job("root")]
        self.assertEqual(plan_jobs(jobs), ["root", "z", "a"])

    def test_multiple_dependencies_and_shared_prerequisite(self):
        jobs = [
            Job("finish", 100, ("left", "right")),
            Job("left", 2, ("root",)),
            Job("right", 1, ("root",)),
            Job("root"),
        ]
        self.assertEqual(plan_jobs(jobs), ["root", "left", "right", "finish"])

    def test_repeated_dependencies_count_once(self):
        jobs = [Job("child", 100, ("root", "root", "other")), Job("root"), Job("other")]
        self.assertEqual(plan_jobs(jobs), ["root", "other", "child"])

    def test_generator_input(self):
        jobs = (job for job in [Job("child", 5, ("root",)), Job("root")])
        self.assertEqual(plan_jobs(jobs), ["root", "child"])

    def test_long_reverse_ordered_chain(self):
        jobs = [
            Job(str(index), depends_on=(str(index - 1),) if index else ())
            for index in reversed(range(1500))
        ]
        self.assertEqual(plan_jobs(jobs), [str(index) for index in range(1500)])

    def test_self_dependency_is_a_cycle(self):
        with self.assertRaisesRegex(ScheduleError, "cycle.*self"):
            plan_jobs([Job("self", depends_on=("self",))])

    def test_cycle_diagnostic_includes_involved_ids(self):
        jobs = [
            Job("alpha", depends_on=("gamma",)),
            Job("beta", depends_on=("alpha",)),
            Job("gamma", depends_on=("beta",)),
        ]
        with self.assertRaises(ScheduleError) as caught:
            plan_jobs(jobs)
        message = str(caught.exception)
        self.assertIn("cycle", message)
        for job in jobs:
            self.assertIn(job.id, message)

    def test_cycle_with_downstream_and_disconnected_jobs(self):
        jobs = [
            Job("ready", 100),
            Job("a", depends_on=("b",)),
            Job("b", depends_on=("a",)),
            Job("downstream", depends_on=("b",)),
        ]
        with self.assertRaises(ScheduleError) as caught:
            plan_jobs(jobs)
        message = str(caught.exception)
        self.assertIn("cycle", message)
        self.assertIn("'a'", message)
        self.assertIn("'b'", message)
        self.assertIn("blocked job IDs", message)
        self.assertNotIn("'ready'", message)

    def test_jobs_dependencies_and_payloads_are_not_mutated(self):
        dependencies = ["root", "root"]
        payload = {"items": [1, 2]}
        jobs = [Job("child", 10, dependencies, payload), Job("root")]
        snapshot = deepcopy(jobs)
        original_jobs = tuple(jobs)
        self.assertEqual(plan_jobs(jobs), ["root", "child"])
        self.assertEqual(plan_jobs(jobs), ["root", "child"])
        self.assertEqual(jobs, snapshot)
        for original, current in zip(original_jobs, jobs):
            self.assertIs(original, current)
        self.assertIs(jobs[0].depends_on, dependencies)
        self.assertIs(jobs[0].payload, payload)

    def test_invalid_input_is_not_mutated(self):
        cases = [
            [Job("a", depends_on=["b"]), Job("b", depends_on=["a"])],
            [Job("a", depends_on=["missing"])],
            [Job("a"), Job("a")],
        ]
        for jobs in cases:
            with self.subTest(jobs=jobs):
                snapshot = deepcopy(jobs)
                with self.assertRaises(ScheduleError):
                    plan_jobs(jobs)
                self.assertEqual(jobs, snapshot)

    def test_small_dags_match_best_valid_permutation(self):
        # An independent oracle: enumerate valid schedules and pick the best
        # lexicographic priority/input-order ranking, without simulating a queue.
        ids = ("a", "b", "c", "d")
        possible_edges = list(combinations(ids, 2))
        input_order = ("d", "b", "a", "c")
        priorities = {"a": 2, "b": -1, "c": 2, "d": 5}
        for mask in range(1 << len(possible_edges)):
            with self.subTest(edge_mask=mask):
                edges = [
                    edge for bit, edge in enumerate(possible_edges)
                    if mask & (1 << bit)
                ]
                jobs = [
                    Job(job_id, priorities[job_id], tuple(
                        source for source, target in edges if target == job_id
                    ))
                    for job_id in input_order
                ]
                valid_orders = [
                    order for order in permutations(ids)
                    if all(order.index(source) < order.index(target) for source, target in edges)
                ]
                expected = min(valid_orders, key=lambda order: [
                    (-priorities[job_id], input_order.index(job_id)) for job_id in order
                ])
                self.assertEqual(plan_jobs(jobs), list(expected))


if __name__ == "__main__":
    unittest.main()

gpt-6.1-sol/01-feature-building/RESPONSE.md

python3 -B -m unittest -v

Ran 24 tests in 0.003s

OK

gpt-6.1-sol/01-feature-building/TASK.md

gpt-6.1-sol/01-feature-building/jobqueue.py

from dataclasses import dataclass, field
from heapq import heappop, heappush
from typing import Any, Iterable


class ScheduleError(ValueError):
    pass


@dataclass(frozen=True)
class Job:
    id: str
    priority: int = 0
    depends_on: tuple[str, ...] = field(default_factory=tuple)
    payload: Any = None


def plan_jobs(jobs: Iterable[Job]) -> list[str]:
    """Order jobs by dependencies, then ready-job priority and input position.

    Repeated dependencies count once. Invalid IDs, missing dependencies, and
    dependency cycles raise ScheduleError without modifying the supplied jobs.
    """
    jobs = list(jobs)
    positions = {}
    for index, job in enumerate(jobs):
        if not isinstance(job.id, str) or not job.id:
            raise ScheduleError(
                f"Job at input position {index} has invalid ID {job.id!r}; "
                "IDs must be non-empty strings"
            )
        if job.id in positions:
            raise ScheduleError(f"Duplicate job ID: {job.id!r}")
        positions[job.id] = index

    remaining = [0] * len(jobs)
    dependents = [[] for _ in jobs]
    for index, job in enumerate(jobs):
        seen = set()
        for dependency in job.depends_on:
            if not isinstance(dependency, str) or not dependency:
                raise ScheduleError(
                    f"Job {job.id!r} has invalid dependency {dependency!r}; "
                    "dependencies must name supplied jobs with non-empty string IDs"
                )
            if dependency not in positions:
                raise ScheduleError(
                    f"Job {job.id!r} has unknown dependency {dependency!r}"
                )
            if dependency not in seen:
                seen.add(dependency)
                remaining[index] += 1
                dependents[positions[dependency]].append(index)

    ready = []
    for index, job in enumerate(jobs):
        if remaining[index] == 0:
            heappush(ready, (-job.priority, index))

    order = []
    while ready:
        _, index = heappop(ready)
        order.append(jobs[index].id)
        for dependent in dependents[index]:
            remaining[dependent] -= 1
            if remaining[dependent] == 0:
                heappush(ready, (-jobs[dependent].priority, dependent))

    if len(order) != len(jobs):
        unresolved = ", ".join(
            repr(job.id) for index, job in enumerate(jobs) if remaining[index]
        )
        raise ScheduleError(
            f"Dependency cycle detected; unresolved jobs: {unresolved}"
        )
    return order

gpt-6.1-sol/01-feature-building/test_jobqueue.py

from copy import deepcopy
import unittest

from jobqueue import Job, ScheduleError, plan_jobs


class PlanJobsTests(unittest.TestCase):
    def test_dependencies_override_priority(self):
        jobs = [
            Job("build", priority=100, depends_on=("fetch",)),
            Job("fetch", priority=1),
        ]
        self.assertEqual(plan_jobs(jobs), ["fetch", "build"])

    def test_ready_jobs_use_priority_then_input_order(self):
        jobs = [Job("a", 2), Job("b", 5), Job("c", 2)]
        self.assertEqual(plan_jobs(jobs), ["b", "a", "c"])

    def test_unknown_dependency_is_rejected(self):
        with self.assertRaises(ScheduleError) as raised:
            plan_jobs([Job("build", depends_on=("missing",))])
        self.assertIn("build", str(raised.exception))
        self.assertIn("missing", str(raised.exception))

    def test_empty_input(self):
        self.assertEqual(plan_jobs([]), [])
        self.assertEqual(plan_jobs(iter(())), [])

    def test_single_job(self):
        self.assertEqual(plan_jobs([Job("only")]), ["only"])

    def test_invalid_ids_are_rejected(self):
        for invalid_id in ("", None, 0, False, [], {}, b"job"):
            with self.subTest(id=invalid_id):
                with self.assertRaises(ScheduleError) as raised:
                    plan_jobs([Job(invalid_id)])
                self.assertIn("non-empty strings", str(raised.exception))
                self.assertIn("position 0", str(raised.exception))

    def test_duplicate_ids_are_rejected(self):
        with self.assertRaises(ScheduleError) as raised:
            plan_jobs([Job("same", 1), Job("same", 2)])
        self.assertIn("Duplicate", str(raised.exception))
        self.assertIn("same", str(raised.exception))

    def test_invalid_dependency_ids_are_rejected(self):
        for dependency in ("", None, 0, False, [], {}, b"source"):
            with self.subTest(dependency=dependency):
                with self.assertRaises(ScheduleError) as raised:
                    plan_jobs([
                        Job("source"),
                        Job("consumer", depends_on=(dependency,)),
                    ])
                self.assertIn("consumer", str(raised.exception))
                self.assertIn("invalid dependency", str(raised.exception))

    def test_valid_ids_need_not_be_identifiers(self):
        jobs = [Job("完成", depends_on=(" ",)), Job(" ")]
        self.assertEqual(plan_jobs(jobs), [" ", "完成"])

    def test_dependencies_can_refer_to_later_input(self):
        jobs = [
            Job("last", depends_on=("middle",)),
            Job("middle", depends_on=("first",)),
            Job("first"),
        ]
        self.assertEqual(plan_jobs(jobs), ["first", "middle", "last"])

    def test_newly_ready_job_can_preempt_existing_ready_jobs(self):
        jobs = [
            Job("waiting", 2),
            Job("unlock", 3),
            Job("urgent", 100, depends_on=("unlock",)),
        ]
        self.assertEqual(plan_jobs(jobs), ["unlock", "urgent", "waiting"])

    def test_newly_ready_tie_uses_original_input_order(self):
        jobs = [
            Job("earlier", 5, depends_on=("unlock",)),
            Job("waiting", 5),
            Job("unlock", 10),
        ]
        self.assertEqual(plan_jobs(jobs), ["unlock", "earlier", "waiting"])

    def test_ties_do_not_use_lexical_id_order(self):
        self.assertEqual(
            plan_jobs([Job("z"), Job("a"), Job("m")]), ["z", "a", "m"]
        )

    def test_negative_priorities(self):
        jobs = [Job("low", -10), Job("high", -1), Job("tied", -1)]
        self.assertEqual(plan_jobs(jobs), ["high", "tied", "low"])

    def test_fan_in_fan_out_and_disconnected_jobs(self):
        jobs = [
            Job("finish", 100, depends_on=("left", "right")),
            Job("right", 2, depends_on=("root",)),
            Job("independent", 0),
            Job("left", 1, depends_on=("root",)),
            Job("root", -1),
        ]
        result = plan_jobs(jobs)
        self.assertEqual(result, ["independent", "root", "right", "left", "finish"])
        self.assertCountEqual(result, [job.id for job in jobs])
        positions = {job_id: index for index, job_id in enumerate(result)}
        for job in jobs:
            for dependency in job.depends_on:
                self.assertLess(positions[dependency], positions[job.id])

    def test_repeated_dependencies_count_once(self):
        jobs = [
            Job("finish", depends_on=("source", "other", "source", "other")),
            Job("source"),
            Job("other"),
        ]
        self.assertEqual(plan_jobs(jobs), ["source", "other", "finish"])

    def test_generator_input(self):
        jobs = (job for job in [Job("b", depends_on=("a",)), Job("a")])
        self.assertEqual(plan_jobs(jobs), ["a", "b"])

    def test_self_dependency_is_a_cycle(self):
        with self.assertRaises(ScheduleError) as raised:
            plan_jobs([Job("self", depends_on=("self",))])
        self.assertIn("cycle", str(raised.exception))
        self.assertIn("self", str(raised.exception))

    def test_two_job_cycle(self):
        with self.assertRaises(ScheduleError) as raised:
            plan_jobs([
                Job("alpha", depends_on=("beta",)),
                Job("beta", depends_on=("alpha",)),
            ])
        self.assertIn("cycle", str(raised.exception))
        self.assertIn("alpha", str(raised.exception))
        self.assertIn("beta", str(raised.exception))

    def test_cycle_with_independent_and_downstream_jobs(self):
        jobs = [
            Job("alpha", depends_on=("gamma",)),
            Job("beta", depends_on=("alpha",)),
            Job("gamma", depends_on=("beta",)),
            Job("blocked", depends_on=("gamma",)),
            Job("independent", 100),
        ]
        with self.assertRaises(ScheduleError) as raised:
            plan_jobs(jobs)
        message = str(raised.exception)
        self.assertIn("cycle", message)
        for job_id in ("alpha", "beta", "gamma", "blocked"):
            self.assertIn(job_id, message)
        self.assertNotIn("independent", message)

    def test_missing_dependency_is_validated_even_in_a_cycle(self):
        with self.assertRaises(ScheduleError) as raised:
            plan_jobs([Job("loop", depends_on=("loop", "missing"))])
        self.assertIn("unknown dependency", str(raised.exception))
        self.assertIn("missing", str(raised.exception))

    def test_success_does_not_mutate_input(self):
        dependencies = ["source", "source"]
        payload = {"nested": [1, 2, 3]}
        jobs = [
            Job("consumer", 100, depends_on=dependencies, payload=payload),
            Job("source"),
        ]
        before = deepcopy(jobs)
        original_jobs = tuple(jobs)
        for _ in range(2):
            self.assertEqual(plan_jobs(jobs), ["source", "consumer"])
            self.assertEqual(jobs, before)
        self.assertTrue(all(job is original for job, original in zip(jobs, original_jobs)))
        self.assertIs(jobs[0].depends_on, dependencies)
        self.assertIs(jobs[0].payload, payload)

    def test_failure_does_not_mutate_input(self):
        for dependencies in (["missing"], ["consumer"]):
            with self.subTest(dependencies=dependencies):
                jobs = [Job("consumer", depends_on=dependencies), Job("source")]
                before = deepcopy(jobs)
                with self.assertRaises(ScheduleError):
                    plan_jobs(jobs)
                self.assertEqual(jobs, before)
                self.assertIs(jobs[0].depends_on, dependencies)

    def test_long_dependency_chain(self):
        jobs = [
            Job(f"job-{index}", depends_on=(f"job-{index - 1}",))
            for index in range(1499, 0, -1)
        ]
        jobs.append(Job("job-0"))
        self.assertEqual(plan_jobs(jobs), [f"job-{index}" for index in range(1500)])


if __name__ == "__main__":
    unittest.main()

mcode-m3.1-flash/01-feature-building/RESPONSE.md

$ python3 -m unittest -v

Ran 44 tests in 0.006s

OK

unknown dep -> job 'build' depends on unknown job 'ghost'; known jobs: build, other
dup id      -> duplicate job id: 'a' appears at positions 0 and 2; job ids must be unique
cycle       -> dependency cycle detected: a -> b -> c -> a; 4 job(s) can never run
               (directly or transitively): a, b, c, downstream
self cycle  -> dependency cycle detected: solo -> solo; 1 job(s) can never run
               (directly or transitively): solo

mcode-m3.1-flash/01-feature-building/TASK.md

mcode-m3.1-flash/01-feature-building/jobqueue.py

"""Dependency-aware, deterministic job scheduling (standard library only).

The public surface is :class:`Job`, :class:`ScheduleError` and :func:`plan_jobs`.
"""

from collections.abc import Iterable
from dataclasses import dataclass, field
import heapq
from typing import Any


class ScheduleError(ValueError):
    pass


@dataclass(frozen=True)
class Job:
    id: str
    priority: int = 0
    depends_on: tuple[str, ...] = field(default_factory=tuple)
    payload: Any = None


_WHITE = 0
_GREY = 1
_BLACK = 2


def _describe_cycle(cycle: list[str]) -> str:
    return " -> ".join(cycle)


def _find_cycle(blocked: set[str], depends: dict[str, tuple[str, ...]]) -> list[str] | None:
    """Return one concrete cycle among `blocked`, or ``None``.

    Only dependencies that are themselves still blocked can close a loop, so
    already-scheduled dependencies are skipped. The walk is iterative on
    purpose: a deep dependency chain must not exhaust the interpreter stack.
    """
    color = {job_id: _WHITE for job_id in blocked}
    for root in blocked:
        if color[root] != _WHITE:
            continue
        color[root] = _GREY
        path: list[str] = [root]
        frames = [iter(dep for dep in depends[root] if dep in blocked)]
        while frames:
            advanced = False
            for dep in frames[-1]:
                state = color[dep]
                if state == _GREY:
                    return path[path.index(dep):] + [dep]
                if state == _WHITE:
                    color[dep] = _GREY
                    path.append(dep)
                    frames.append(iter(d for d in depends[dep] if d in blocked))
                    advanced = True
                    break
            if not advanced:
                frames.pop()
                color[path.pop()] = _BLACK
    return None


def plan_jobs(jobs: Iterable[Job]) -> list[str]:
    """Return job IDs in a valid deterministic execution order.

    A job is emitted only after every job it depends on, and among the jobs
    that are ready at any moment the highest ``priority`` goes first, with
    equal priorities resolved by original input order. ``ScheduleError`` is
    raised for malformed input (bad or duplicated ids, unknown dependencies)
    and for dependency cycles. The supplied jobs are never mutated.
    """
    try:
        materialised = list(jobs)
    except TypeError:
        raise ScheduleError(f"jobs must be an iterable of Job instances, got {jobs!r}") from None

    if not materialised:
        return []

    # Pass 1 -- index every job and reject malformed or duplicated entries.
    ids: list[str] = []
    position: dict[str, int] = {}
    priority: dict[str, float] = {}
    depends: dict[str, tuple[str, ...]] = {}

    for index, job in enumerate(materialised):
        if not isinstance(job, Job):
            raise ScheduleError(f"item at position {index} is not a Job: {job!r}")

        job_id = job.id
        if not isinstance(job_id, str):
            raise ScheduleError(
                f"job at position {index} has a non-string id: {job_id!r}"
            )
        if not job_id.strip():
            raise ScheduleError(
                f"job at position {index} has a blank id; job ids must be non-empty strings"
            )
        if job_id in position:
            raise ScheduleError(
                f"duplicate job id: {job_id!r} appears at positions "
                f"{position[job_id]} and {index}; job ids must be unique"
            )
        if not isinstance(job.priority, (int, float)):
            raise ScheduleError(
                f"job {job_id!r} has a non-numeric priority: {job.priority!r}"
            )

        raw_deps = job.depends_on
        if isinstance(raw_deps, str) or not isinstance(raw_deps, Iterable):
            raise ScheduleError(
                f"job {job_id!r} has an invalid depends_on: {raw_deps!r}; "
                "expected an iterable of job id strings"
            )
        # Deduplicate while preserving order so a repeated dependency does not
        # inflate the dependency count and stall the job forever.
        unique_deps: list[str] = []
        seen: set[str] = set()
        for dep in raw_deps:
            if not isinstance(dep, str) or not dep:
                raise ScheduleError(
                    f"job {job_id!r} has an invalid dependency: {dep!r}; "
                    "dependencies must be non-empty job id strings"
                )
            if dep not in seen:
                seen.add(dep)
                unique_deps.append(dep)

        position[job_id] = index
        ids.append(job_id)
        priority[job_id] = job.priority
        depends[job_id] = tuple(unique_deps)

    # Pass 2 -- every dependency must name a job that was actually supplied.
    dependents: dict[str, list[str]] = {job_id: [] for job_id in ids}
    indegree: dict[str, int] = {}
    for job_id in ids:
        pending = 0
        for dep in depends[job_id]:
            if dep not in position:
                raise ScheduleError(
                    f"job {job_id!r} depends on unknown job {dep!r}; "
                    f"known jobs: {', '.join(ids)}"
                )
            dependents[dep].append(job_id)
            pending += 1
        indegree[job_id] = pending

    # Kahn's algorithm: repeatedly release the best currently-ready job.
    # The heap key is (-priority, input position): priority first, then the
    # original input order as a stable tie-break.
    ready: list[tuple[float, int]] = [
        (-priority[job_id], position[job_id]) for job_id in ids if indegree[job_id] == 0
    ]
    heapq.heapify(ready)

    order: list[str] = []
    scheduled: set[str] = set()
    while ready:
        _, index = heapq.heappop(ready)
        job_id = ids[index]
        if job_id in scheduled:  # defensive: heap keys are unique
            continue
        scheduled.add(job_id)
        order.append(job_id)
        for dependent in dependents[job_id]:
            indegree[dependent] -= 1
            if indegree[dependent] == 0:
                heapq.heappush(
                    ready, (-priority[dependent], position[dependent])
                )

    if len(order) != len(ids):
        blocked = {job_id for job_id in ids if job_id not in scheduled}
        cycle = _find_cycle(blocked, depends)
        headline = (
            f"dependency cycle detected: {_describe_cycle(cycle)}"
            if cycle
            else "unable to schedule: dependency graph has no complete ordering"
        )
        raise ScheduleError(
            f"{headline}; {len(blocked)} job(s) can never run "
            f"(directly or transitively): {', '.join(sorted(blocked))}"
        )

    return order

mcode-m3.1-flash/01-feature-building/test_jobqueue.py

import copy
import unittest

from jobqueue import Job, ScheduleError, plan_jobs


def assert_valid_order(testcase, jobs, order):
    """Assert the contract properties hold for `order` over `jobs`."""
    expected = [job.id for job in jobs]
    testcase.assertEqual(sorted(order), sorted(expected), "every job id exactly once")
    testcase.assertEqual(len(order), len(set(order)), "no duplicates")

    seen = []
    for job_id in order:
        job = next(j for j in jobs if j.id == job_id)
        for dep in job.depends_on:
            testcase.assertIn(
                dep, seen, f"{job_id!r} emitted before its dependency {dep!r}"
            )
        seen.append(job_id)


class PlanJobsTests(unittest.TestCase):
    def test_dependencies_override_priority(self):
        jobs = [
            Job("build", priority=100, depends_on=("fetch",)),
            Job("fetch", priority=1),
        ]
        self.assertEqual(plan_jobs(jobs), ["fetch", "build"])

    def test_ready_jobs_use_priority_then_input_order(self):
        jobs = [Job("a", 2), Job("b", 5), Job("c", 2)]
        self.assertEqual(plan_jobs(jobs), ["b", "a", "c"])

    def test_unknown_dependency_is_rejected(self):
        with self.assertRaises(ScheduleError):
            plan_jobs([Job("build", depends_on=("missing",))])

    # --- baseline / trivial input ---------------------------------------

    def test_empty_input_returns_empty_list(self):
        self.assertEqual(plan_jobs([]), [])

    def test_single_job(self):
        self.assertEqual(plan_jobs([Job("only")]), ["only"])

    def test_no_dependencies_preserves_input_order(self):
        jobs = [Job("a"), Job("b"), Job("c")]
        self.assertEqual(plan_jobs(jobs), ["a", "b", "c"])

    def test_accepts_any_iterable_including_a_generator(self):
        jobs = (Job("a", 1, depends_on=("b",)), Job("b", 2))
        self.assertEqual(plan_jobs(iter(jobs)), ["b", "a"])

    def test_generator_is_consumed_only_once(self):
        gen = (job for job in [Job("a", 5), Job("b", 1)])
        self.assertEqual(plan_jobs(gen), ["a", "b"])
        self.assertEqual(list(gen), [], "the generator is not replayed")

    def test_schedule_error_is_a_value_error(self):
        self.assertTrue(issubclass(ScheduleError, ValueError))

    # --- id validation ---------------------------------------------------

    def test_duplicate_id_is_rejected(self):
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs([Job("dup"), Job("other"), Job("dup")])
        self.assertIn("dup", str(ctx.exception))

    def test_empty_id_is_rejected(self):
        with self.assertRaises(ScheduleError):
            plan_jobs([Job(""), Job("b")])

    def test_whitespace_only_id_is_rejected(self):
        with self.assertRaises(ScheduleError):
            plan_jobs([Job("   ")])

    def test_non_string_id_is_rejected(self):
        with self.assertRaises(ScheduleError):
            plan_jobs([Job(7)])

    def test_non_string_dependency_is_rejected(self):
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs([Job("a", depends_on=(3,))])
        self.assertIn("a", str(ctx.exception))

    def test_bare_string_depends_on_is_rejected(self):
        # A plain string is iterable but would silently split into characters.
        with self.assertRaises(ScheduleError):
            plan_jobs([Job("a", depends_on="b"), Job("b")])  # type: ignore[arg-type]

    def test_none_depends_on_is_rejected(self):
        with self.assertRaises(ScheduleError):
            plan_jobs([Job("a", depends_on=None)])  # type: ignore[arg-type]

    def test_non_job_item_is_rejected(self):
        with self.assertRaises(ScheduleError):
            plan_jobs(["not-a-job"])  # type: ignore[list-item]

    def test_non_numeric_priority_is_rejected(self):
        with self.assertRaises(ScheduleError):
            plan_jobs([Job("a", priority="high")])  # type: ignore[arg-type]

    def test_non_iterable_input_is_rejected(self):
        with self.assertRaises(ScheduleError):
            plan_jobs(42)  # type: ignore[arg-type]

    # --- dependency resolution -------------------------------------------

    def test_unknown_dependency_message_names_the_missing_job(self):
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs([Job("build", depends_on=("ghost",))])
        message = str(ctx.exception)
        self.assertIn("ghost", message)
        self.assertIn("build", message)

    def test_declaration_order_does_not_matter(self):
        jobs = [
            Job("last", depends_on=("middle",)),
            Job("middle", depends_on=("first",)),
            Job("first"),
        ]
        self.assertEqual(plan_jobs(jobs), ["first", "middle", "last"])

    def test_duplicate_dependency_entries_do_not_stall(self):
        jobs = [Job("a", depends_on=("b", "b", "b")), Job("b")]
        self.assertEqual(plan_jobs(jobs), ["b", "a"])

    def test_depends_on_may_be_a_list_and_is_not_mutated(self):
        deps = ["b", "c"]
        jobs = [Job("a", depends_on=deps), Job("b"), Job("c")]
        # 'a' is blocked; {b, c} are ready and tie on priority, so they keep
        # input order and 'a' comes last.
        self.assertEqual(plan_jobs(jobs), ["b", "c", "a"])
        self.assertEqual(deps, ["b", "c"], "the caller's list is untouched")

    def test_diamond_dependency(self):
        #   top -> left -> base, top -> right -> base
        jobs = [
            Job("top", 0, depends_on=("left", "right")),
            Job("left", 5, depends_on=("base",)),
            Job("right", 1, depends_on=("base",)),
            Job("base", 9),
        ]
        order = plan_jobs(jobs)
        # base(9) is the only ready job; it unblocks left(5) and right(1),
        # so left wins on priority; 'top' waits for both.
        self.assertEqual(order, ["base", "left", "right", "top"])
        # 'top' is not released until both 'left' and 'right' are done.
        self.assertLess(order.index("left"), order.index("top"))
        self.assertLess(order.index("right"), order.index("top"))
        assert_valid_order(self, jobs, order)

    def test_priority_only_applies_among_ready_jobs(self):
        jobs = [
            Job("low_prereq", 0),
            Job("high_blocked", 100, depends_on=("low_prereq",)),
            Job("mid_free", 50),
        ]
        # At t=0 the ready set is {low_prereq(0), mid_free(50)}; mid_free wins.
        # Only then does high_blocked(100) become runnable -- a blocked
        # high-priority job never preempts an already-ready one.
        self.assertEqual(plan_jobs(jobs), ["mid_free", "low_prereq", "high_blocked"])

    def test_equal_priorities_fall_back_to_input_order(self):
        jobs = [Job("j%d" % i, 1) for i in range(6)]
        self.assertEqual(plan_jobs(jobs), [job.id for job in jobs])

    def test_negative_and_zero_priorities(self):
        jobs = [Job("neg", -5), Job("zero", 0), Job("pos", 5)]
        self.assertEqual(plan_jobs(jobs), ["pos", "zero", "neg"])

    def test_is_deterministic_across_repeated_calls(self):
        jobs = [
            Job("a", 3, depends_on=("c",)),
            Job("b", 3, depends_on=("a",)),
            Job("c", 1),
            Job("d", 3),
        ]
        first = plan_jobs(jobs)
        for _ in range(5):
            self.assertEqual(plan_jobs(jobs), first)

    def test_wide_fan_in(self):
        jobs = [Job("leaf%d" % i) for i in range(50)]
        jobs.append(Job("fan", 1, depends_on=tuple("leaf%d" % i for i in range(50))))
        order = plan_jobs(jobs)
        self.assertEqual(order[-1], "fan")
        assert_valid_order(self, jobs, order)

    def test_deep_chain_does_not_exhaust_the_stack(self):
        depth = 2000
        jobs = [Job("step0")]
        jobs += [Job("step%d" % i, depends_on=("step%d" % (i - 1),)) for i in range(1, depth)]
        order = plan_jobs(jobs)
        self.assertEqual(len(order), depth)
        self.assertEqual(order[0], "step0")
        self.assertEqual(order[-1], "step%d" % (depth - 1))

    def test_every_id_returned_exactly_once(self):
        jobs = [
            Job("a", 1, depends_on=("b", "c")),
            Job("b", 2, depends_on=("d",)),
            Job("c", 3),
            Job("d", 0),
            Job("e", 4, depends_on=("a",)),
        ]
        order = plan_jobs(jobs)
        self.assertEqual(sorted(order), ["a", "b", "c", "d", "e"])
        assert_valid_order(self, jobs, order)

    # --- cycle detection -------------------------------------------------

    def test_self_dependency_is_a_cycle(self):
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs([Job("loop", depends_on=("loop",))])
        self.assertIn("loop", str(ctx.exception))

    def test_two_node_cycle(self):
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs([Job("a", depends_on=("b",)), Job("b", depends_on=("a",))])
        self.assertIn("a", str(ctx.exception))
        self.assertIn("b", str(ctx.exception))

    def test_three_node_cycle_names_every_member(self):
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs([
                Job("a", depends_on=("c",)),
                Job("b", depends_on=("a",)),
                Job("c", depends_on=("b",)),
            ])
        message = str(ctx.exception)
        for job_id in ("a", "b", "c"):
            self.assertIn(job_id, message)

    def test_cycle_message_identifies_the_cycle_not_unrelated_jobs(self):
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs([
                Job("loopy", depends_on=("other",)),
                Job("other", depends_on=("loopy",)),
                Job("innocent", 5),
            ])
        message = str(ctx.exception)
        self.assertIn("cycle", message)
        self.assertIn("loopy", message)
        self.assertIn("other", message)
        self.assertNotIn("innocent", message, "unrelated runnable job is not a casualty")

    def test_jobs_blocked_by_a_cycle_are_reported_separately(self):
        # 'downstream' is not itself in a cycle, but can never run either.
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs([
                Job("a", depends_on=("b",)),
                Job("b", depends_on=("a",)),
                Job("downstream", depends_on=("a",)),
            ])
        message = str(ctx.exception)
        self.assertIn("cycle", message)
        self.assertIn("downstream", message)

    def test_long_cycle_is_reported(self):
        size = 60
        jobs = [
            Job("n%d" % i, depends_on=("n%d" % ((i + 1) % size),)) for i in range(size)
        ]
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs(jobs)
        self.assertIn("n0", str(ctx.exception))
        self.assertIn("n%d" % (size - 1), str(ctx.exception))

    def test_cycle_fails_the_whole_plan(self):
        # Even though most jobs are schedulable, a cycle aborts everything.
        with self.assertRaises(ScheduleError):
            plan_jobs([Job("ok"), Job("a", depends_on=("b",)), Job("b", depends_on=("a",))])

    def test_deep_cycle_does_not_exhaust_the_stack(self):
        depth = 2000
        jobs = [Job("c0", depends_on=("c%d" % (depth - 1),))]
        jobs += [Job("c%d" % i, depends_on=("c%d" % (i - 1),)) for i in range(1, depth)]
        with self.assertRaises(ScheduleError):
            plan_jobs(jobs)

    # --- immutability ----------------------------------------------------

    def test_does_not_mutate_the_supplied_jobs(self):
        jobs = [
            Job("build", 2, depends_on=("fetch", "compile")),
            Job("fetch", 1),
            Job("compile", 3),
        ]
        before = copy.deepcopy(jobs)
        plan_jobs(jobs)
        self.assertEqual(jobs, before, "job objects are unchanged")

    def test_does_not_mutate_dependency_lists(self):
        deps = ["b"]
        jobs = [Job("a", 1, depends_on=deps), Job("b", 2)]
        plan_jobs(jobs)
        self.assertEqual(deps, ["b"])

    def test_does_not_mutate_on_failure_paths(self):
        jobs = [Job("a", depends_on=("b",)), Job("b", depends_on=("a",))]
        before = copy.deepcopy(jobs)
        with self.assertRaises(ScheduleError):
            plan_jobs(jobs)
        self.assertEqual(jobs, before)

    def test_result_is_a_fresh_list_each_call(self):
        jobs = [Job("a"), Job("b")]
        first = plan_jobs(jobs)
        first.append("tampered")
        self.assertEqual(plan_jobs(jobs), ["a", "b"])

    def test_does_not_mutate_payloads(self):
        payload = {"touches": [1, 2, 3]}
        jobs = [Job("a", 1, payload=payload), Job("b", 2)]
        plan_jobs(jobs)
        self.assertEqual(payload, {"touches": [1, 2, 3]})


if __name__ == "__main__":
    unittest.main()

gpt-6-luna/01-feature-building/RESPONSE.md

python3 -m unittest -v

gpt-6-luna/01-feature-building/TASK.md

gpt-6-luna/01-feature-building/jobqueue.py

from dataclasses import dataclass, field
import heapq
from typing import Any, Iterable


class ScheduleError(ValueError):
    pass


@dataclass(frozen=True)
class Job:
    id: str
    priority: int = 0
    depends_on: tuple[str, ...] = field(default_factory=tuple)
    payload: Any = None


def plan_jobs(jobs: Iterable[Job]) -> list[str]:
    """Return job IDs in a valid deterministic execution order."""
    supplied_jobs = list(jobs)

    # Keep the input order in the index: it is the stable tie breaker among
    # jobs with equal priority.
    jobs_by_id: dict[str, Job] = {}
    for job in supplied_jobs:
        if not isinstance(job, Job):
            raise ScheduleError(f"Expected a Job instance, got {type(job).__name__}")
        if not isinstance(job.id, str) or not job.id:
            raise ScheduleError("Job IDs must be non-empty strings")
        if job.id in jobs_by_id:
            raise ScheduleError(f"Duplicate job ID: {job.id!r}")
        jobs_by_id[job.id] = job

    # Build reverse edges so that completing a prerequisite releases its
    # dependent jobs. Copy dependencies into local structures only.
    dependents: dict[str, list[str]] = {job_id: [] for job_id in jobs_by_id}
    remaining_dependencies: dict[str, int] = {}
    for job in supplied_jobs:
        try:
            dependencies = tuple(job.depends_on)
        except TypeError as exc:
            raise ScheduleError(f"Dependencies for job {job.id!r} must be iterable") from exc

        for dependency in dependencies:
            if not isinstance(dependency, str) or not dependency:
                raise ScheduleError(
                    f"Dependencies for job {job.id!r} must be non-empty job ID strings"
                )
            if dependency not in jobs_by_id:
                raise ScheduleError(
                    f"Job {job.id!r} depends on unknown job {dependency!r}"
                )
            dependents[dependency].append(job.id)
        remaining_dependencies[job.id] = len(dependencies)

    job_indices = {job.id: index for index, job in enumerate(supplied_jobs)}
    ready: list[tuple[int, int, str]] = []
    for job in supplied_jobs:
        if remaining_dependencies[job.id] == 0:
            try:
                heapq.heappush(ready, (-job.priority, job_indices[job.id], job.id))
            except TypeError as exc:
                raise ScheduleError(
                    f"Priority for job {job.id!r} must be an ordered numeric value"
                ) from exc

    order: list[str] = []
    while ready:
        _, _, job_id = heapq.heappop(ready)
        order.append(job_id)

        for dependent_id in dependents[job_id]:
            remaining_dependencies[dependent_id] -= 1
            if remaining_dependencies[dependent_id] == 0:
                dependent = jobs_by_id[dependent_id]
                try:
                    heapq.heappush(
                        ready,
                        (-dependent.priority, job_indices[dependent_id], dependent_id),
                    )
                except TypeError as exc:
                    raise ScheduleError(
                        f"Priority for job {dependent_id!r} must be an ordered numeric value"
                    ) from exc

    if len(order) != len(supplied_jobs):
        unscheduled = [job.id for job in supplied_jobs if remaining_dependencies[job.id] > 0]
        raise ScheduleError(
            "Dependency cycle detected; jobs still blocked: " + ", ".join(unscheduled)
        )

    return order

gpt-6-luna/01-feature-building/test_jobqueue.py

import unittest

from jobqueue import Job, ScheduleError, plan_jobs


class PlanJobsTests(unittest.TestCase):
    def test_empty_input(self):
        self.assertEqual(plan_jobs([]), [])

    def test_dependencies_override_priority(self):
        jobs = [
            Job("build", priority=100, depends_on=("fetch",)),
            Job("fetch", priority=1),
        ]
        self.assertEqual(plan_jobs(jobs), ["fetch", "build"])

    def test_ready_jobs_use_priority_then_input_order(self):
        jobs = [Job("a", 2), Job("b", 5), Job("c", 2)]
        self.assertEqual(plan_jobs(jobs), ["b", "a", "c"])

    def test_unknown_dependency_is_rejected(self):
        with self.assertRaisesRegex(ScheduleError, "unknown job 'missing'"):
            plan_jobs([Job("build", depends_on=("missing",))])

    def test_newly_ready_high_priority_job_precedes_waiting_jobs(self):
        jobs = [
            Job("first", priority=10),
            Job("second", priority=1),
            Job("urgent", priority=20, depends_on=("first",)),
        ]
        self.assertEqual(plan_jobs(jobs), ["first", "urgent", "second"])

    def test_equal_priority_jobs_keep_input_order_when_released_together(self):
        jobs = [
            Job("dependent-later", priority=3, depends_on=("root",)),
            Job("root", priority=4),
            Job("already-ready", priority=3),
        ]
        self.assertEqual(plan_jobs(jobs), ["root", "dependent-later", "already-ready"])

    def test_rejects_empty_and_non_string_ids(self):
        for job_id in ("", None, 12):
            with self.subTest(job_id=job_id):
                with self.assertRaisesRegex(ScheduleError, "non-empty strings"):
                    plan_jobs([Job(job_id)])

    def test_rejects_duplicate_ids(self):
        with self.assertRaisesRegex(ScheduleError, "Duplicate job ID: 'same'"):
            plan_jobs([Job("same"), Job("same")])

    def test_rejects_invalid_dependencies(self):
        for dependency in ("", None, 42):
            with self.subTest(dependency=dependency):
                with self.assertRaisesRegex(ScheduleError, "non-empty job ID strings"):
                    plan_jobs([Job("build", depends_on=(dependency,))])

    def test_self_dependency_is_a_cycle(self):
        with self.assertRaisesRegex(ScheduleError, "cycle.*self"):
            plan_jobs([Job("self", depends_on=("self",))])

    def test_multi_job_cycle_reports_blocked_ids(self):
        jobs = [
            Job("a", depends_on=("b",)),
            Job("b", depends_on=("a",)),
            Job("blocked", depends_on=("a",)),
        ]
        with self.assertRaises(ScheduleError) as raised:
            plan_jobs(jobs)
        self.assertIn("a", str(raised.exception))
        self.assertIn("b", str(raised.exception))

    def test_does_not_mutate_jobs_or_dependency_lists(self):
        dependencies = ["root"]
        jobs = [Job("child", depends_on=dependencies), Job("root")]
        original_jobs = jobs.copy()

        self.assertEqual(plan_jobs(jobs), ["root", "child"])
        self.assertEqual(jobs, original_jobs)
        self.assertEqual(dependencies, ["root"])
        self.assertIs(jobs[0].depends_on, dependencies)


if __name__ == "__main__":
    unittest.main()

muse/01-feature-building/RESPONSE.md

muse/01-feature-building/TASK.md

muse/01-feature-building/jobqueue.py

from dataclasses import dataclass, field
from typing import Any, Iterable


class ScheduleError(ValueError):
    pass


@dataclass(frozen=True)
class Job:
    id: str
    priority: int = 0
    depends_on: tuple[str, ...] = field(default_factory=tuple)
    payload: Any = None


def plan_jobs(jobs: Iterable[Job]) -> list[str]:
    """Return job IDs in a valid deterministic execution order."""
    if jobs is None:
        raise ScheduleError("jobs must be an iterable of Job, got None")
    try:
        job_list = list(jobs)
    except TypeError as exc:
        raise ScheduleError(f"jobs must be an iterable of Job: {exc}") from exc
    if not job_list:
        return []

    id_to_job: dict[str, Job] = {}
    id_to_index: dict[str, int] = {}
    id_to_priority: dict[str, int] = {}
    for index, job in enumerate(job_list):
        if not isinstance(job, Job):
            raise ScheduleError(f"item at index {index} is not a Job: {job!r}")
        jid = job.id
        if not isinstance(jid, str) or not jid:
            raise ScheduleError(
                f"job at index {index} has invalid id {jid!r}: must be a non-empty string"
            )
        if jid in id_to_job:
            raise ScheduleError(
                f"duplicate job id {jid!r} at index {index} "
                f"(first seen at index {id_to_index[jid]})"
            )
        id_to_job[jid] = job
        id_to_index[jid] = index
        id_to_priority[jid] = job.priority

    indegree: dict[str, int] = {jid: 0 for jid in id_to_job}
    dependents: dict[str, list[str]] = {jid: [] for jid in id_to_job}
    for job in job_list:
        raw = job.depends_on
        if raw is None:
            raise ScheduleError(
                f"job {job.id!r} has invalid depends_on None: must be an iterable of job IDs"
            )
        if isinstance(raw, (str, bytes)):
            raise ScheduleError(
                f"job {job.id!r} has invalid depends_on {raw!r}: "
                "must be an iterable of job IDs, not a string"
            )
        try:
            dep_list = list(raw)
        except TypeError as exc:
            raise ScheduleError(
                f"job {job.id!r} has invalid depends_on {raw!r}: must be an iterable of job IDs"
            ) from exc
        seen: set[str] = set()
        for dep in dep_list:
            if not isinstance(dep, str) or not dep:
                raise ScheduleError(
                    f"job {job.id!r} has invalid dependency {dep!r}: must be a non-empty string"
                )
            if dep not in id_to_job:
                raise ScheduleError(f"job {job.id!r} depends on unknown job {dep!r}")
            if dep in seen:
                continue
            seen.add(dep)
            indegree[job.id] += 1
            dependents[dep].append(job.id)

    ready = [jid for jid, deg in indegree.items() if deg == 0]
    result: list[str] = []
    while ready:
        best = max(ready, key=lambda jid: (id_to_priority[jid], -id_to_index[jid]))
        ready.remove(best)
        result.append(best)
        for child in dependents[best]:
            indegree[child] -= 1
            if indegree[child] == 0:
                ready.append(child)

    if len(result) != len(job_list):
        remaining = sorted(
            (jid for jid, deg in indegree.items() if deg > 0),
            key=lambda jid: id_to_index[jid],
        )
        detail = ", ".join(repr(j) for j in remaining)
        raise ScheduleError(f"dependency cycle detected involving: {detail}")

    return result

muse/01-feature-building/test_jobqueue.py

import unittest

from jobqueue import Job, ScheduleError, plan_jobs


class PlanJobsTests(unittest.TestCase):
    def test_dependencies_override_priority(self):
        jobs = [
            Job("build", priority=100, depends_on=("fetch",)),
            Job("fetch", priority=1),
        ]
        self.assertEqual(plan_jobs(jobs), ["fetch", "build"])

    def test_ready_jobs_use_priority_then_input_order(self):
        jobs = [Job("a", 2), Job("b", 5), Job("c", 2)]
        self.assertEqual(plan_jobs(jobs), ["b", "a", "c"])

    def test_unknown_dependency_is_rejected(self):
        with self.assertRaises(ScheduleError):
            plan_jobs([Job("build", depends_on=("missing",))])

    def test_empty_input_returns_empty_list(self):
        self.assertEqual(plan_jobs([]), [])
        self.assertEqual(plan_jobs(()), [])

    def test_single_job(self):
        self.assertEqual(plan_jobs([Job("solo", priority=3)]), ["solo"])

    def test_duplicate_ids_rejected(self):
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs([Job("a"), Job("b"), Job("a")])
        self.assertIn("a", str(ctx.exception))

    def test_empty_string_id_rejected(self):
        with self.assertRaises(ScheduleError):
            plan_jobs([Job("")])

    def test_non_string_id_rejected(self):
        with self.assertRaises(ScheduleError):
            plan_jobs([Job(123)])  # type: ignore[arg-type]
        with self.assertRaises(ScheduleError):
            plan_jobs([Job(None)])  # type: ignore[arg-type]

    def test_unknown_dependency_message_is_useful(self):
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs([Job("build", depends_on=("missing",))])
        msg = str(ctx.exception)
        self.assertIn("build", msg)
        self.assertIn("missing", msg)

    def test_invalid_dependency_entries_rejected(self):
        with self.assertRaises(ScheduleError):
            plan_jobs([Job("a", depends_on=("",))])
        with self.assertRaises(ScheduleError):
            plan_jobs([Job("a", depends_on=(123,))])  # type: ignore[arg-type]
        with self.assertRaises(ScheduleError):
            plan_jobs([Job("a", depends_on=None)])  # type: ignore[arg-type]

    def test_string_depends_on_rejected(self):
        # A bare string must not be treated as an iterable of IDs.
        with self.assertRaises(ScheduleError):
            plan_jobs([Job("a", depends_on="b"), Job("b")])  # type: ignore[arg-type]

    def test_self_dependency_is_cycle(self):
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs([Job("a", depends_on=("a",))])
        self.assertIn("a", str(ctx.exception))

    def test_direct_cycle_reports_ids(self):
        jobs = [Job("a", depends_on=("b",)), Job("b", depends_on=("a",))]
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs(jobs)
        msg = str(ctx.exception)
        self.assertIn("a", msg)
        self.assertIn("b", msg)

    def test_longer_cycle_reports_ids(self):
        jobs = [
            Job("a", depends_on=("c",)),
            Job("b", depends_on=("a",)),
            Job("c", depends_on=("b",)),
            Job("ok"),
        ]
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs(jobs)
        msg = str(ctx.exception)
        for jid in ("a", "b", "c"):
            self.assertIn(jid, msg)

    def test_duplicate_dependencies_tolerated(self):
        jobs = [Job("a", depends_on=("b", "b")), Job("b")]
        self.assertEqual(plan_jobs(jobs), ["b", "a"])

    def test_diamond_dependencies(self):
        jobs = [
            Job("a"),
            Job("b", depends_on=("a",)),
            Job("c", depends_on=("a",)),
            Job("d", depends_on=("b", "c")),
        ]
        self.assertEqual(plan_jobs(jobs), ["a", "b", "c", "d"])

    def test_chain_overrides_priority(self):
        jobs = [
            Job("c", priority=100, depends_on=("b",)),
            Job("b", priority=50, depends_on=("a",)),
            Job("a", priority=1),
        ]
        self.assertEqual(plan_jobs(jobs), ["a", "b", "c"])

    def test_newly_ready_high_priority_jumps_ahead(self):
        jobs = [
            Job("low", priority=1),
            Job("blocked_high", priority=100, depends_on=("low",)),
            Job("mid", priority=50),
        ]
        self.assertEqual(plan_jobs(jobs), ["mid", "low", "blocked_high"])

    def test_equal_priority_keeps_input_order_after_unblock(self):
        jobs = [
            Job("root"),
            Job("x", priority=5, depends_on=("root",)),
            Job("y", priority=5, depends_on=("root",)),
        ]
        self.assertEqual(plan_jobs(jobs), ["root", "x", "y"])

    def test_does_not_mutate_inputs(self):
        jobs = [Job("b", priority=2), Job("a", priority=1, depends_on=("b",))]
        snapshot = [(j.id, j.priority, tuple(j.depends_on)) for j in jobs]
        plan_jobs(jobs)
        self.assertEqual([(j.id, j.priority, tuple(j.depends_on)) for j in jobs], snapshot)

    def test_does_not_mutate_list_depends_on(self):
        deps = ["y"]
        jobs = [Job("x", depends_on=deps), Job("y")]  # type: ignore[arg-type]
        self.assertEqual(plan_jobs(jobs), ["y", "x"])
        self.assertEqual(deps, ["y"])

    def test_accepts_generator_input(self):
        def gen():
            yield Job("b", priority=1)
            yield Job("a", priority=9, depends_on=("b",))

        self.assertEqual(plan_jobs(gen()), ["b", "a"])

    def test_returns_every_id_exactly_once(self):
        jobs = [
            Job("a", priority=2),
            Job("b", priority=3, depends_on=("a",)),
            Job("c", priority=3),
            Job("d", priority=1, depends_on=("b", "c")),
            Job("e", priority=9, depends_on=("d",)),
        ]
        result = plan_jobs(jobs)
        self.assertEqual(sorted(result), ["a", "b", "c", "d", "e"])
        pos = {jid: i for i, jid in enumerate(result)}
        for job in jobs:
            for dep in job.depends_on:
                self.assertLess(pos[dep], pos[job.id])


if __name__ == "__main__":
    unittest.main()

mimo/01-feature-building/RESPONSE.md

python3 -m unittest test_jobqueue -v

Ran 21 tests in 0.001s

OK

mimo/01-feature-building/TASK.md

mimo/01-feature-building/jobqueue.py

from __future__ import annotations

import heapq
from dataclasses import dataclass, field
from typing import Any, Iterable


class ScheduleError(ValueError):
    pass


@dataclass(frozen=True)
class Job:
    id: str
    priority: int = 0
    depends_on: tuple[str, ...] = field(default_factory=tuple)
    payload: Any = None


def plan_jobs(jobs: Iterable[Job]) -> list[str]:
    """Return job IDs in a valid deterministic execution order.

    Among ready jobs, higher ``priority`` runs first; ties keep input order.
    """
    job_list = list(jobs)
    if not job_list:
        return []

    by_id: dict[str, Job] = {}
    input_index: dict[str, int] = {}
    for index, job in enumerate(job_list):
        job_id = job.id
        if not isinstance(job_id, str) or not job_id:
            raise ScheduleError(f"job id must be a non-empty string, got {job_id!r}")
        if job_id in by_id:
            raise ScheduleError(f"duplicate job id: {job_id!r}")
        by_id[job_id] = job
        input_index[job_id] = index

    # dependents[d] = jobs that list d as a dependency; remaining counts unmet deps.
    dependents: dict[str, list[str]] = {job_id: [] for job_id in by_id}
    remaining: dict[str, int] = {job_id: 0 for job_id in by_id}

    for job in job_list:
        # Copy so a caller-supplied mutable sequence is never touched.
        seen: set[str] = set()
        for dep in tuple(job.depends_on):
            if dep not in by_id:
                raise ScheduleError(f"job {job.id!r} depends on unknown job {dep!r}")
            if dep in seen:
                continue
            seen.add(dep)
            dependents[dep].append(job.id)
            remaining[job.id] += 1

    # Kahn's algorithm; ready set ordered by (-priority, input_index, id).
    ready: list[tuple[int, int, str]] = []
    for job_id, unmet in remaining.items():
        if unmet == 0:
            job = by_id[job_id]
            heapq.heappush(ready, (-job.priority, input_index[job_id], job_id))

    order: list[str] = []
    while ready:
        _, _, job_id = heapq.heappop(ready)
        order.append(job_id)
        for dependent in dependents[job_id]:
            remaining[dependent] -= 1
            if remaining[dependent] == 0:
                job = by_id[dependent]
                heapq.heappush(
                    ready, (-job.priority, input_index[dependent], dependent)
                )

    if len(order) != len(by_id):
        stuck = sorted(job_id for job_id, unmet in remaining.items() if unmet > 0)
        cycle = _find_cycle(stuck, by_id)
        detail = " -> ".join(cycle) if cycle else ", ".join(stuck)
        raise ScheduleError(
            f"dependency cycle detected: {detail}"
            f" (unresolvable jobs: {', '.join(stuck)})"
        )

    return order


def _find_cycle(stuck: list[str], by_id: dict[str, Job]) -> list[str]:
    """Return one dependency cycle as [id, ..., id], or [] if none found."""
    stuck_set = set(stuck)
    color = dict.fromkeys(stuck, 0)  # 0=unvisited, 1=on stack, 2=done

    def dfs(node: str, path: list[str]) -> list[str]:
        color[node] = 1
        path.append(node)
        for dep in dict.fromkeys(by_id[node].depends_on):
            if dep not in stuck_set:
                continue
            if color[dep] == 1:
                start = path.index(dep)
                return path[start:]
            if color[dep] == 0:
                found = dfs(dep, path)
                if found:
                    return found
        path.pop()
        color[node] = 2
        return []

    for job_id in stuck:
        if color[job_id] == 0:
            found = dfs(job_id, [])
            if found:
                return found
    return []

mimo/01-feature-building/test_jobqueue.py

import unittest

from jobqueue import Job, ScheduleError, plan_jobs


class PlanJobsTests(unittest.TestCase):
    def test_dependencies_override_priority(self):
        jobs = [
            Job("build", priority=100, depends_on=("fetch",)),
            Job("fetch", priority=1),
        ]
        self.assertEqual(plan_jobs(jobs), ["fetch", "build"])

    def test_ready_jobs_use_priority_then_input_order(self):
        jobs = [Job("a", 2), Job("b", 5), Job("c", 2)]
        self.assertEqual(plan_jobs(jobs), ["b", "a", "c"])

    def test_unknown_dependency_is_rejected(self):
        with self.assertRaises(ScheduleError):
            plan_jobs([Job("build", depends_on=("missing",))])

    def test_empty_input(self):
        self.assertEqual(plan_jobs([]), [])

    def test_single_job(self):
        self.assertEqual(plan_jobs([Job("only")]), ["only"])

    def test_all_ids_returned_exactly_once(self):
        jobs = [
            Job("a", depends_on=("b", "c")),
            Job("b", depends_on=("d",)),
            Job("c", depends_on=("d",)),
            Job("d"),
        ]
        order = plan_jobs(jobs)
        self.assertEqual(sorted(order), ["a", "b", "c", "d"])
        self.assertEqual(len(order), 4)

    def test_diamond_dependencies(self):
        jobs = [
            Job("top", depends_on=("left", "right")),
            Job("left", depends_on=("root",)),
            Job("right", depends_on=("root",)),
            Job("root"),
        ]
        order = plan_jobs(jobs)
        self.assertEqual(order[0], "root")
        self.assertEqual(order[-1], "top")
        self.assertLess(order.index("left"), order.index("top"))
        self.assertLess(order.index("right"), order.index("top"))

    def test_priority_applies_when_job_becomes_ready(self):
        jobs = [
            Job("x", priority=1, depends_on=("z",)),
            Job("y", priority=2),
            Job("z", priority=0),
        ]
        self.assertEqual(plan_jobs(jobs), ["y", "z", "x"])

    def test_equal_priority_keeps_input_order_for_late_ready_jobs(self):
        jobs = [
            Job("late1", priority=3, depends_on=("root",)),
            Job("late2", priority=3, depends_on=("root",)),
            Job("root", priority=0),
        ]
        self.assertEqual(plan_jobs(jobs), ["root", "late1", "late2"])

    def test_negative_priority_orders_last_among_ready(self):
        jobs = [Job("low", priority=-5), Job("high", priority=0)]
        self.assertEqual(plan_jobs(jobs), ["high", "low"])

    def test_duplicate_job_id_rejected(self):
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs([Job("a"), Job("a")])
        self.assertIn("duplicate", str(ctx.exception).lower())

    def test_empty_job_id_rejected(self):
        with self.assertRaises(ScheduleError):
            plan_jobs([Job("")])

    def test_non_string_job_id_rejected(self):
        with self.assertRaises(ScheduleError):
            plan_jobs([Job(123)])  # type: ignore[arg-type]

    def test_unknown_dependency_message_names_jobs(self):
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs([Job("build", depends_on=("missing",))])
        message = str(ctx.exception)
        self.assertIn("build", message)
        self.assertIn("missing", message)

    def test_self_dependency_is_a_cycle(self):
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs([Job("a", depends_on=("a",))])
        self.assertIn("a", str(ctx.exception))

    def test_two_job_cycle(self):
        jobs = [
            Job("a", depends_on=("b",)),
            Job("b", depends_on=("a",)),
        ]
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs(jobs)
        message = str(ctx.exception)
        self.assertIn("cycle", message.lower())
        self.assertIn("a", message)
        self.assertIn("b", message)

    def test_longer_cycle_reports_involved_ids(self):
        jobs = [
            Job("a", depends_on=("b",)),
            Job("b", depends_on=("c",)),
            Job("c", depends_on=("a",)),
            Job("ok"),
        ]
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs(jobs)
        message = str(ctx.exception)
        for job_id in ("a", "b", "c"):
            self.assertIn(job_id, message)

    def test_cycle_with_bystander_dependency(self):
        jobs = [
            Job("a", depends_on=("b",)),
            Job("b", depends_on=("a",)),
            Job("c", depends_on=("a",)),
        ]
        with self.assertRaises(ScheduleError) as ctx:
            plan_jobs(jobs)
        message = str(ctx.exception)
        self.assertIn("a", message)
        self.assertIn("b", message)

    def test_duplicate_entries_in_depends_on(self):
        jobs = [
            Job("build", depends_on=("fetch", "fetch")),
            Job("fetch"),
        ]
        self.assertEqual(plan_jobs(jobs), ["fetch", "build"])

    def test_does_not_mutate_jobs_or_dependencies(self):
        deps = ["fetch"]
        jobs = [Job("build", depends_on=tuple(deps)), Job("fetch", priority=1)]
        snapshot = [(job.id, job.priority, job.depends_on) for job in jobs]
        plan_jobs(jobs)
        self.assertEqual(
            [(job.id, job.priority, job.depends_on) for job in jobs], snapshot
        )
        self.assertEqual(deps, ["fetch"])

    def test_generator_input(self):
        def gen():
            yield Job("b", priority=1, depends_on=("a",))
            yield Job("a", priority=9)

        self.assertEqual(plan_jobs(gen()), ["a", "b"])


if __name__ == "__main__":
    unittest.main()

mcode-m3/01-feature-building/RESPONSE.md

python3 -m unittest test_jobqueue.py -v

Ran 20 tests in 0.000s

OK

mcode-m3/01-feature-building/TASK.md

mcode-m3/01-feature-building/jobqueue.py

from dataclasses import dataclass, field
from typing import Any, Iterable
import heapq


class ScheduleError(ValueError):
    """Raised when jobs cannot be scheduled due to invalid input or cycles."""


@dataclass(frozen=True)
class Job:
    id: str
    priority: int = 0
    depends_on: tuple[str, ...] = field(default_factory=tuple)
    payload: Any = None


def plan_jobs(jobs: Iterable[Job]) -> list[str]:
    """Return job IDs in a valid deterministic execution order.

    Validation
    ----------
    - Every ``Job.id`` must be a non-empty string and unique across the
      input. Violations raise :class:`ScheduleError` and identify the
      offending value and its position.
    - Every entry in ``depends_on`` must name another job in the input.
      Unknown references raise :class:`ScheduleError` identifying both
      the dependent and the missing job.

    Scheduling
    ----------
    - The result lists every supplied id exactly once.
    - A job is only emitted once every job it depends on has already been
      emitted.
    - Among jobs that are currently ready (no outstanding dependencies),
      the one with the highest ``priority`` is emitted next; ties are
      broken by original input order (earlier in the iterable first).
    - If a dependency cycle is detected, :class:`ScheduleError` is raised
      and the ids of jobs still holding unsatisfied dependencies are
      included in the message.

    The supplied jobs and their ``depends_on`` tuples are never mutated.
    """
    # One pass over the iterable so we can both validate ids and remember
    # the original input position used to break priority ties.
    job_list = list(jobs)

    # --- Validate ids ------------------------------------------------------
    seen: set[str] = set()
    for index, job in enumerate(job_list):
        if not isinstance(job.id, str) or not job.id:
            raise ScheduleError(
                f"Job at position {index} has invalid id {job.id!r}; "
                "ids must be non-empty strings"
            )
        if job.id in seen:
            raise ScheduleError(
                f"Duplicate job id {job.id!r} at position {index}"
            )
        seen.add(job.id)

    if not job_list:
        return []

    id_to_job = {job.id: job for job in job_list}
    order_index = {job.id: index for index, job in enumerate(job_list)}

    # --- Validate dependencies --------------------------------------------
    for job in job_list:
        for dep in job.depends_on:
            if dep not in id_to_job:
                raise ScheduleError(
                    f"Job {job.id!r} depends on unknown job {dep!r}"
                )

    # --- Build the graph --------------------------------------------------
    # dependents[dep_id] is the list of jobs that depend on dep_id.
    # pending[job_id] is the SET of dep_ids for job_id that have not yet
    # been emitted. We copy the depends_on tuple into a fresh set so the
    # caller's tuple is never touched.
    dependents: dict[str, list[str]] = {job.id: [] for job in job_list}
    pending: dict[str, set[str]] = {
        job.id: set(job.depends_on) for job in job_list
    }
    for job in job_list:
        for dep in job.depends_on:
            dependents[dep].append(job.id)

    # --- Schedule --------------------------------------------------------
    # Min-heap keyed on (-priority, input_order, id). heapq gives us the
    # smallest key first, so negating priority puts higher priorities
    # first; ties fall back to the original input position.
    ready: list[tuple[int, int, str]] = []
    for job in job_list:
        if not pending[job.id]:
            heapq.heappush(
                ready,
                (-job.priority, order_index[job.id], job.id),
            )

    result: list[str] = []
    while ready:
        _, _, job_id = heapq.heappop(ready)
        result.append(job_id)
        for dependent_id in dependents[job_id]:
            pending[dependent_id].discard(job_id)
            if not pending[dependent_id]:
                dependent_job = id_to_job[dependent_id]
                heapq.heappush(
                    ready,
                    (
                        -dependent_job.priority,
                        order_index[dependent_id],
                        dependent_id,
                    ),
                )

    if len(result) != len(job_list):
        # Whatever is still in `pending` belongs to one or more cycles.
        cyclic = sorted(job_id for job_id, deps in pending.items() if deps)
        raise ScheduleError(
            f"Dependency cycle detected involving jobs: {cyclic}"
        )

    return result

mcode-m3/01-feature-building/test_jobqueue.py

import unittest

from jobqueue import Job, ScheduleError, plan_jobs


class PlanJobsTests(unittest.TestCase):
    def test_dependencies_override_priority(self):
        jobs = [
            Job("build", priority=100, depends_on=("fetch",)),
            Job("fetch", priority=1),
        ]
        self.assertEqual(plan_jobs(jobs), ["fetch", "build"])

    def test_ready_jobs_use_priority_then_input_order(self):
        jobs = [Job("a", 2), Job("b", 5), Job("c", 2)]
        self.assertEqual(plan_jobs(jobs), ["b", "a", "c"])

    def test_unknown_dependency_is_rejected(self):
        with self.assertRaises(ScheduleError):
            plan_jobs([Job("build", depends_on=("missing",))])

    # --- Trivial input -----------------------------------------------------

    def test_empty_input_returns_empty_list(self):
        self.assertEqual(plan_jobs([]), [])

    def test_single_job_with_no_deps(self):
        self.assertEqual(plan_jobs([Job("solo")]), ["solo"])

    def test_generator_input_is_consumed_once(self):
        def gen():
            yield Job("a", priority=1)
            yield Job("b", priority=2)

        self.assertEqual(plan_jobs(gen()), ["b", "a"])

    # --- Shape coverage ----------------------------------------------------

    def test_diamond_dependency_preserves_partial_order(self):
        # A is depended on by B and D; C depends on both B and D.
        jobs = [
            Job("A"),
            Job("B", depends_on=("A",)),
            Job("D", depends_on=("A",)),
            Job("C", depends_on=("B", "D")),
        ]
        result = plan_jobs(jobs)
        self.assertEqual(len(result), 4)
        self.assertEqual(set(result), {"A", "B", "D", "C"})
        idx = {jid: i for i, jid in enumerate(result)}
        self.assertLess(idx["A"], idx["B"])
        self.assertLess(idx["A"], idx["D"])
        self.assertLess(idx["B"], idx["C"])
        self.assertLess(idx["D"], idx["C"])

    def test_complex_chain_with_priorities(self):
        # Highest priority sits in the middle of the chain; correctness of
        # the topological order must override the priority preference.
        jobs = [
            Job("compile", priority=1),
            Job("link", priority=10, depends_on=("compile",)),
            Job("test", priority=5, depends_on=("link",)),
            Job("package", priority=5, depends_on=("test",)),
        ]
        self.assertEqual(
            plan_jobs(jobs),
            ["compile", "link", "test", "package"],
        )

    def test_ready_jobs_with_equal_priority_use_input_order(self):
        jobs = [Job("first"), Job("b"), Job("c"), Job("last")]
        self.assertEqual(plan_jobs(jobs), ["first", "b", "c", "last"])

    def test_disjoint_subgraphs_are_ordered_independently(self):
        jobs = [
            Job("x1"),
            Job("x2", depends_on=("x1",)),
            Job("y1"),
            Job("y2", depends_on=("y1",)),
        ]
        result = plan_jobs(jobs)
        idx = {jid: i for i, jid in enumerate(result)}
        self.assertEqual(set(result), {"x1", "x2", "y1", "y2"})
        self.assertLess(idx["x1"], idx["x2"])
        self.assertLess(idx["y1"], idx["y2"])

    def test_result_is_complete_and_unique(self):
        jobs = [
            Job("a"),
            Job("b", depends_on=("a",)),
            Job("c", depends_on=("a", "b")),
            Job("d", depends_on=("c",)),
            Job("e", depends_on=("c",)),
        ]
        result = plan_jobs(jobs)
        self.assertEqual(len(result), len(set(result)))
        self.assertEqual(set(result), {"a", "b", "c", "d", "e"})

    # --- Priorities --------------------------------------------------------

    def test_negative_priorities_order_lower_first(self):
        jobs = [Job("a", priority=-5), Job("b", priority=-1)]
        self.assertEqual(plan_jobs(jobs), ["b", "a"])

    # --- Cycle detection ---------------------------------------------------

    def test_cycle_of_two_jobs_raises_with_ids_in_message(self):
        jobs = [
            Job("a", depends_on=("b",)),
            Job("b", depends_on=("a",)),
        ]
        with self.assertRaises(ScheduleError) as cm:
            plan_jobs(jobs)
        msg = str(cm.exception)
        self.assertIn("'a'", msg)
        self.assertIn("'b'", msg)

    def test_cycle_of_three_jobs_raises_with_ids_in_message(self):
        jobs = [
            Job("a", depends_on=("c",)),
            Job("b", depends_on=("a",)),
            Job("c", depends_on=("b",)),
        ]
        with self.assertRaises(ScheduleError) as cm:
            plan_jobs(jobs)
        msg = str(cm.exception)
        for jid in ("a", "b", "c"):
            self.assertIn(f"'{jid}'", msg)

    def test_self_loop_is_a_cycle(self):
        with self.assertRaises(ScheduleError):
            plan_jobs([Job("a", depends_on=("a",))])

    def test_cycle_diagnostic_excludes_unrelated_jobs(self):
        # "free" is independent of the a<->b cycle.
        jobs = [
            Job("free"),
            Job("a", depends_on=("b",)),
            Job("b", depends_on=("a",)),
        ]
        with self.assertRaises(ScheduleError) as cm:
            plan_jobs(jobs)
        msg = str(cm.exception)
        self.assertIn("'a'", msg)
        self.assertIn("'b'", msg)
        # The unrelated job must not be blamed.
        self.assertNotIn("'free'", msg)

    # --- Input validation --------------------------------------------------

    def test_duplicate_id_is_rejected(self):
        jobs = [Job("dup", priority=1), Job("dup", priority=2)]
        with self.assertRaises(ScheduleError) as cm:
            plan_jobs(jobs)
        self.assertIn("dup", str(cm.exception))

    def test_empty_id_string_is_rejected(self):
        with self.assertRaises(ScheduleError):
            plan_jobs([Job("")])

    def test_non_string_id_is_rejected(self):
        with self.assertRaises(ScheduleError):
            plan_jobs([Job(123)])  # type: ignore[arg-type]
        with self.assertRaises(ScheduleError):
            plan_jobs([Job(None)])  # type: ignore[arg-type]

    # --- Non-mutation ------------------------------------------------------

    def test_inputs_are_not_mutated(self):
        jobs = [
            Job("a", priority=3, depends_on=()),
            Job("b", priority=1, depends_on=("a",)),
            Job("c", priority=2, depends_on=("b",)),
        ]
        snapshot_ids = [job.id for job in jobs]
        snapshot_deps = [job.depends_on for job in jobs]
        snapshot_priorities = [job.priority for job in jobs]
        snapshot_payloads = [job.payload for job in jobs]

        plan_jobs(jobs)

        self.assertEqual([job.id for job in jobs], snapshot_ids)
        self.assertEqual([job.depends_on for job in jobs], snapshot_deps)
        self.assertEqual([job.priority for job in jobs], snapshot_priorities)
        self.assertEqual([job.payload for job in jobs], snapshot_payloads)


if __name__ == "__main__":
    unittest.main()