From fd9d3900d9db2c311a7fd4b9cc37035357795e2c Mon Sep 17 00:00:00 2001
From: TohruSu <238259854+TohruSu@users.noreply.github.com>
Date: Thu, 15 Oct 2026 11:09:09 +0000
Subject: [PATCH 1/2] Implement discrete-time simulator and scheduling
 baselines

---
 .github/workflows/checks.yml    | 20 ++++++++
 README.md                       | 40 ++++++++++++++-
 docs/simulation.md              | 86 +++++++++++++++++++++++++++++++
 pyproject.toml                  | 36 +++++++++++++
 src/adaptivequeue/__init__.py   | 20 ++++++++
 src/adaptivequeue/__main__.py   |  3 ++
 src/adaptivequeue/cli.py        | 47 +++++++++++++++++
 src/adaptivequeue/metrics.py    | 37 ++++++++++++++
 src/adaptivequeue/models.py     | 45 +++++++++++++++++
 src/adaptivequeue/policies.py   | 39 ++++++++++++++
 src/adaptivequeue/simulation.py | 44 ++++++++++++++++
 src/adaptivequeue/workloads.py  | 90 +++++++++++++++++++++++++++++++++
 tests/test_cli.py               | 43 ++++++++++++++++
 tests/test_metrics.py           | 44 ++++++++++++++++
 tests/test_policies.py          | 51 +++++++++++++++++++
 tests/test_simulation.py        | 66 ++++++++++++++++++++++++
 tests/test_workloads.py         | 69 +++++++++++++++++++++++++
 17 files changed, 778 insertions(+), 2 deletions(-)
 create mode 100644 .github/workflows/checks.yml
 create mode 100644 docs/simulation.md
 create mode 100644 pyproject.toml
 create mode 100644 src/adaptivequeue/__init__.py
 create mode 100644 src/adaptivequeue/__main__.py
 create mode 100644 src/adaptivequeue/cli.py
 create mode 100644 src/adaptivequeue/metrics.py
 create mode 100644 src/adaptivequeue/models.py
 create mode 100644 src/adaptivequeue/policies.py
 create mode 100644 src/adaptivequeue/simulation.py
 create mode 100644 src/adaptivequeue/workloads.py
 create mode 100644 tests/test_cli.py
 create mode 100644 tests/test_metrics.py
 create mode 100644 tests/test_policies.py
 create mode 100644 tests/test_simulation.py
 create mode 100644 tests/test_workloads.py

diff --git a/.github/workflows/checks.yml b/.github/workflows/checks.yml
new file mode 100644
index 0000000..abdee7d
--- /dev/null
+++ b/.github/workflows/checks.yml
@@ -0,0 +1,20 @@
+name: Checks
+
+on: [push, pull_request]
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.10", "3.12"]
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python-version }}
+      - run: python -m pip install -e '.[dev]'
+      - run: python -m pytest
+      - run: ruff check .
+      - run: ruff format --check .
+      - run: mypy
diff --git a/README.md b/README.md
index 2b639da..9438a31 100644
--- a/README.md
+++ b/README.md
@@ -1,2 +1,38 @@
-# adaptivequeue
-A lightweight framework for reinforcement learning-based adaptive job scheduling.
+# AdaptiveQueue
+
+A lightweight Python project for experimenting with adaptive background job
+scheduling. The first milestone provides a discrete-time, single-worker simulator
+and FIFO, SPT, EDF, Priority, and seeded Random baselines.
+
+Requires Python 3.10 or newer. There are no runtime dependencies.
+
+```sh
+python -m venv .venv
+source .venv/bin/activate
+python -m pip install -e '.[dev]'
+```
+
+Run an episode with 60 jobs:
+
+```sh
+adaptivequeue --workload mixed --policy edf --seed 42
+```
+
+The CLI reports priority-weighted scheduling cost (lower is better), deadline miss
+rate, average and 95th-percentile waiting time, and average flow time. Workloads
+are `stable`, `bursty`, or `mixed`; policies are `fifo`, `spt`, `edf`, `priority`,
+or `random`. Use the same seed and workload settings to compare policies fairly.
+
+```sh
+python -m adaptivequeue --workload bursty --policy spt --seed 7 \
+  --jobs 60 --arrival-rate 0.3 --deadline-factor 3
+python -m pytest
+ruff check .
+ruff format --check .
+mypy
+```
+
+See [simulation rules and parameters](docs/simulation.md) or run
+`adaptivequeue --help`. The Python API also exposes per-job execution records.
+Reinforcement learning is planned for a later milestone; the current scope is
+simulation and conventional scheduling baselines.
diff --git a/docs/simulation.md b/docs/simulation.md
new file mode 100644
index 0000000..efb31a8
--- /dev/null
+++ b/docs/simulation.md
@@ -0,0 +1,86 @@
+# Simulation
+
+## Rules
+
+Time starts at zero and uses integer units. A job specifies a unique string
+`job_id`, nonnegative `arrival_time`, positive `duration`, nonnegative absolute
+`deadline`, positive `priority_weight`, and descriptive `job_type`. An already
+unachievable deadline is allowed. Specifications are immutable, so one workload
+can be reused across policies.
+
+One worker runs one job to completion without preemption. Whenever it is idle,
+all jobs with arrival time at or before the current time enter the waiting queue.
+The policy receives only an immutable snapshot of that queue and must select a
+waiting job. It cannot observe future arrivals. Jobs arriving during execution
+wait until completion; arrivals exactly at completion participate in the next
+decision. The engine skips to completion or the next arrival rather than stepping
+through ticks where no scheduling decision is possible. It runs until every job
+finishes, including late jobs, without deliberately idling when work is available.
+
+## Policies
+
+| CLI name | Selects |
+| --- | --- |
+| `fifo` | Earliest arrival |
+| `spt` | Shortest processing duration |
+| `edf` | Earliest absolute deadline |
+| `priority` | Highest priority weight |
+| `random` | Uniformly random waiting job, using a local seeded RNG |
+
+The first four break ties by earliest arrival, then lexicographically smallest
+job ID. Random's queue is populated in arrival/ID order, making seeded results
+independent of input ordering. Recreate a Random policy with the same seed for
+each replay. Workload generation and policy selection use separate RNGs and do
+not modify Python's global random state.
+
+## Workloads
+
+Every generator takes an explicit seed and an optional `WorkloadConfig`.
+
+| Parameter | Default | Meaning |
+| --- | --- | --- |
+| `num_jobs` / `--jobs` | 60 | Jobs in an episode; zero gives an empty episode |
+| `arrival_rate` / `--arrival-rate` | 0.28 | Target mean jobs per time unit |
+| `deadline_factor` / `--deadline-factor` | 4.0 | Relative deadline budget; larger is looser |
+| `duration_range` (Python API) | (1, 6) | Inclusive integer duration bounds |
+| `burst_size` / `--burst-size` | 6 | Jobs per burst, with a smaller final burst if needed |
+| `burst_window` / `--burst-window` | 3 | Burst offsets range from 0 to window minus 1 |
+
+`stable` uses independent exponential interarrival gaps with a constant rate and
+the first arrival at zero. Cumulative arrival times are rounded down to integers,
+so several jobs can arrive together. `bursty` places batches within short windows;
+burst starts have exponential gaps with mean `burst_size / arrival_rate`. Bursts
+can overlap. `mixed` uses stable arrivals with interactive jobs (weight 4, deadline
+multiplier 0.5), reports (weight 2, multiplier 1), and maintenance jobs (weight 1,
+multiplier 2), chosen with equal probability.
+
+Durations are uniform within the configured range. Stable and bursty workloads
+choose weights uniformly from 1, 2, and 4, and descriptive types from email,
+image, and backup. All deadline budgets are
+`max(1, ceil(duration * deadline_factor * type_multiplier * U(0.75, 1.25)))`;
+the absolute deadline is arrival plus budget. Stable and bursty type multipliers
+are 1. Increasing arrival rate increases contention. The default rate gives
+nominal utilization `0.28 * 3.5 = 0.98`; finite episodes vary by seed.
+
+## Records and evaluation
+
+`simulate(jobs, policy)` returns `SimulationResult`. Its `jobs` tuple holds
+`JobResult` records in execution order, including each job, start time, completion
+time, and derived waiting time, flow time, and tardiness:
+
+- Waiting time = start minus arrival.
+- Flow time = completion minus arrival.
+- Tardiness = max(0, completion minus absolute deadline).
+- Primary cost `J = mean(priority_weight * (tardiness + 0.1 * flow_time))`.
+- Deadline miss rate = fraction of jobs with positive tardiness. Completion on
+  the deadline counts as on time.
+
+Lower primary cost is better. The result also reports average waiting time,
+95th-percentile waiting time, and average flow time. The percentile linearly
+interpolates sorted waiting times at index `0.95 * (n - 1)` (zero-based), and
+equals the lone value for one job. Every aggregate metric is zero for an empty
+episode. Cost is averaged over jobs, without normalizing by total priority weight.
+
+Compare policies on identical generated jobs and settings. Seeds control both
+workloads and Random decisions independently; several episode seeds give a more
+useful comparison than a single run.
diff --git a/pyproject.toml b/pyproject.toml
new file mode 100644
index 0000000..9d3b9a2
--- /dev/null
+++ b/pyproject.toml
@@ -0,0 +1,36 @@
+[build-system]
+requires = ["setuptools>=68"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "adaptivequeue"
+version = "0.1.0"
+description = "A discrete-time job scheduling simulator and baseline policies"
+readme = "README.md"
+requires-python = ">=3.10"
+license = {file = "LICENSE"}
+dependencies = []
+
+[project.optional-dependencies]
+dev = ["pytest>=8", "ruff>=0.9", "mypy>=1.14"]
+
+[project.scripts]
+adaptivequeue = "adaptivequeue.cli:main"
+
+[tool.setuptools.packages.find]
+where = ["src"]
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+
+[tool.ruff]
+target-version = "py310"
+line-length = 100
+
+[tool.ruff.lint]
+select = ["E", "F", "I"]
+
+[tool.mypy]
+files = ["src/adaptivequeue"]
+python_version = "3.10"
+strict = true
diff --git a/src/adaptivequeue/__init__.py b/src/adaptivequeue/__init__.py
new file mode 100644
index 0000000..0b735f9
--- /dev/null
+++ b/src/adaptivequeue/__init__.py
@@ -0,0 +1,20 @@
+"""Discrete-time simulation and conventional job scheduling baselines."""
+
+from adaptivequeue.metrics import Metrics
+from adaptivequeue.models import Job, JobResult
+from adaptivequeue.policies import make_policy
+from adaptivequeue.simulation import SimulationResult, simulate
+from adaptivequeue.workloads import WorkloadConfig, bursty_arrivals, mixed_workload, stable_arrivals
+
+__all__ = [
+    "Job",
+    "JobResult",
+    "Metrics",
+    "SimulationResult",
+    "WorkloadConfig",
+    "bursty_arrivals",
+    "make_policy",
+    "mixed_workload",
+    "simulate",
+    "stable_arrivals",
+]
diff --git a/src/adaptivequeue/__main__.py b/src/adaptivequeue/__main__.py
new file mode 100644
index 0000000..841ab14
--- /dev/null
+++ b/src/adaptivequeue/__main__.py
@@ -0,0 +1,3 @@
+from adaptivequeue.cli import main
+
+main()
diff --git a/src/adaptivequeue/cli.py b/src/adaptivequeue/cli.py
new file mode 100644
index 0000000..d94a6f9
--- /dev/null
+++ b/src/adaptivequeue/cli.py
@@ -0,0 +1,47 @@
+"""Command-line entry point for one reproducible scheduling episode."""
+
+import argparse
+
+from adaptivequeue.policies import POLICY_NAMES, make_policy
+from adaptivequeue.simulation import simulate
+from adaptivequeue.workloads import WORKLOADS, WorkloadConfig
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Run an AdaptiveQueue scheduling episode.")
+    parser.add_argument("--workload", choices=WORKLOADS, default="stable")
+    parser.add_argument("--policy", choices=POLICY_NAMES, default="fifo")
+    parser.add_argument("--seed", type=int, default=42)
+    parser.add_argument("--jobs", type=int, default=60)
+    parser.add_argument("--arrival-rate", type=float, default=0.28, help="Mean jobs per time unit")
+    parser.add_argument(
+        "--deadline-factor", type=float, default=4.0, help="Larger means looser deadlines"
+    )
+    parser.add_argument("--burst-size", type=int, default=6)
+    parser.add_argument(
+        "--burst-window", type=int, default=3, help="Burst arrival window in time units"
+    )
+    args = parser.parse_args()
+
+    try:
+        config = WorkloadConfig(
+            num_jobs=args.jobs,
+            arrival_rate=args.arrival_rate,
+            deadline_factor=args.deadline_factor,
+            burst_size=args.burst_size,
+            burst_window=args.burst_window,
+        )
+    except ValueError as error:
+        parser.error(str(error))
+
+    jobs = WORKLOADS[args.workload](args.seed, config)
+    result = simulate(jobs, make_policy(args.policy, seed=args.seed))
+    metrics = result.metrics
+    print(
+        f"Workload: {args.workload} | Policy: {args.policy} | Seed: {args.seed} | Jobs: {len(jobs)}"
+    )
+    print(f"Primary cost J (lower is better): {metrics.primary_cost:.3f}")
+    print(f"Deadline miss rate: {metrics.deadline_miss_rate:.1%}")
+    print(f"Average waiting time: {metrics.average_waiting_time:.3f}")
+    print(f"95th-percentile waiting time: {metrics.p95_waiting_time:.3f}")
+    print(f"Average flow time: {metrics.average_flow_time:.3f}")
diff --git a/src/adaptivequeue/metrics.py b/src/adaptivequeue/metrics.py
new file mode 100644
index 0000000..f87e39b
--- /dev/null
+++ b/src/adaptivequeue/metrics.py
@@ -0,0 +1,37 @@
+"""Episode metrics; the waiting-time percentile uses linear interpolation."""
+
+from dataclasses import dataclass
+from math import ceil, floor
+from statistics import fmean
+from typing import Sequence
+
+from adaptivequeue.models import JobResult
+
+
+@dataclass(frozen=True)
+class Metrics:
+    primary_cost: float
+    deadline_miss_rate: float
+    average_waiting_time: float
+    p95_waiting_time: float
+    average_flow_time: float
+
+
+def calculate_metrics(results: Sequence[JobResult]) -> Metrics:
+    if not results:
+        return Metrics(0.0, 0.0, 0.0, 0.0, 0.0)
+
+    waiting = sorted(result.waiting_time for result in results)
+    position = 0.95 * (len(waiting) - 1)
+    lower, upper = floor(position), ceil(position)
+    p95 = waiting[lower] + (waiting[upper] - waiting[lower]) * (position - lower)
+    return Metrics(
+        primary_cost=fmean(
+            result.job.priority_weight * (result.tardiness + 0.1 * result.flow_time)
+            for result in results
+        ),
+        deadline_miss_rate=fmean(result.tardiness > 0 for result in results),
+        average_waiting_time=fmean(waiting),
+        p95_waiting_time=p95,
+        average_flow_time=fmean(result.flow_time for result in results),
+    )
diff --git a/src/adaptivequeue/models.py b/src/adaptivequeue/models.py
new file mode 100644
index 0000000..eb5b577
--- /dev/null
+++ b/src/adaptivequeue/models.py
@@ -0,0 +1,45 @@
+"""Immutable job specifications and execution records."""
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class Job:
+    job_id: str
+    arrival_time: int
+    duration: int
+    deadline: int
+    priority_weight: int
+    job_type: str
+
+    def __post_init__(self) -> None:
+        if not self.job_id or not self.job_type:
+            raise ValueError("job_id and job_type must be nonempty")
+        if any(
+            not isinstance(value, int)
+            for value in (self.arrival_time, self.duration, self.deadline, self.priority_weight)
+        ):
+            raise ValueError("job times, duration, and priority_weight must be integers")
+        if self.arrival_time < 0 or self.deadline < 0:
+            raise ValueError("arrival_time and deadline must be nonnegative")
+        if self.duration <= 0 or self.priority_weight <= 0:
+            raise ValueError("duration and priority_weight must be positive")
+
+
+@dataclass(frozen=True)
+class JobResult:
+    job: Job
+    start_time: int
+    completion_time: int
+
+    @property
+    def waiting_time(self) -> int:
+        return self.start_time - self.job.arrival_time
+
+    @property
+    def flow_time(self) -> int:
+        return self.completion_time - self.job.arrival_time
+
+    @property
+    def tardiness(self) -> int:
+        return max(0, self.completion_time - self.job.deadline)
diff --git a/src/adaptivequeue/policies.py b/src/adaptivequeue/policies.py
new file mode 100644
index 0000000..343db64
--- /dev/null
+++ b/src/adaptivequeue/policies.py
@@ -0,0 +1,39 @@
+"""Policies receive only the jobs currently waiting for the worker."""
+
+import random
+from typing import Callable, Sequence
+
+from adaptivequeue.models import Job
+
+Policy = Callable[[Sequence[Job]], Job]
+
+
+def fifo(waiting: Sequence[Job]) -> Job:
+    return min(waiting, key=lambda job: (job.arrival_time, job.job_id))
+
+
+def spt(waiting: Sequence[Job]) -> Job:
+    return min(waiting, key=lambda job: (job.duration, job.arrival_time, job.job_id))
+
+
+def edf(waiting: Sequence[Job]) -> Job:
+    return min(waiting, key=lambda job: (job.deadline, job.arrival_time, job.job_id))
+
+
+def priority(waiting: Sequence[Job]) -> Job:
+    return min(waiting, key=lambda job: (-job.priority_weight, job.arrival_time, job.job_id))
+
+
+POLICIES: dict[str, Policy] = {"fifo": fifo, "spt": spt, "edf": edf, "priority": priority}
+POLICY_NAMES = (*POLICIES, "random")
+
+
+def make_policy(name: str, *, seed: int) -> Policy:
+    """Create a policy; recreate it with the same seed to replay a random run."""
+    if name == "random":
+        rng = random.Random(seed)
+        return rng.choice
+    try:
+        return POLICIES[name]
+    except KeyError:
+        raise ValueError(f"unknown policy: {name}") from None
diff --git a/src/adaptivequeue/simulation.py b/src/adaptivequeue/simulation.py
new file mode 100644
index 0000000..7ff5e3a
--- /dev/null
+++ b/src/adaptivequeue/simulation.py
@@ -0,0 +1,44 @@
+"""A single-worker, non-preemptive simulation on integer time boundaries."""
+
+from dataclasses import dataclass
+from typing import Iterable
+
+from adaptivequeue.metrics import Metrics, calculate_metrics
+from adaptivequeue.models import Job, JobResult
+from adaptivequeue.policies import Policy
+
+
+@dataclass(frozen=True)
+class SimulationResult:
+    jobs: tuple[JobResult, ...]
+    metrics: Metrics
+
+
+def simulate(jobs: Iterable[Job], policy: Policy) -> SimulationResult:
+    """Run until every job finishes, returning records in execution order."""
+    arrivals = sorted(jobs, key=lambda job: (job.arrival_time, job.job_id))
+    if len({job.job_id for job in arrivals}) != len(arrivals):
+        raise ValueError("job IDs must be unique within an episode")
+
+    waiting: list[Job] = []
+    completed: list[JobResult] = []
+    time = 0
+    next_arrival = 0
+
+    while waiting or next_arrival < len(arrivals):
+        if not waiting:
+            time = max(time, arrivals[next_arrival].arrival_time)
+        while next_arrival < len(arrivals) and arrivals[next_arrival].arrival_time <= time:
+            waiting.append(arrivals[next_arrival])
+            next_arrival += 1
+
+        selected = policy(tuple(waiting))
+        if selected not in waiting:
+            raise ValueError("policy must select a job from the waiting queue")
+        waiting.remove(selected)
+        completion_time = time + selected.duration
+        completed.append(JobResult(selected, time, completion_time))
+        time = completion_time
+
+    records = tuple(completed)
+    return SimulationResult(records, calculate_metrics(records))
diff --git a/src/adaptivequeue/workloads.py b/src/adaptivequeue/workloads.py
new file mode 100644
index 0000000..c0b421d
--- /dev/null
+++ b/src/adaptivequeue/workloads.py
@@ -0,0 +1,90 @@
+"""Seeded synthetic workloads with configurable load and deadline budgets."""
+
+import random
+from dataclasses import dataclass
+from math import ceil, isfinite
+from typing import Callable
+
+from adaptivequeue.models import Job
+
+
+@dataclass(frozen=True)
+class WorkloadConfig:
+    num_jobs: int = 60
+    arrival_rate: float = 0.28
+    deadline_factor: float = 4.0
+    duration_range: tuple[int, int] = (1, 6)
+    burst_size: int = 6
+    burst_window: int = 3
+
+    def __post_init__(self) -> None:
+        if self.num_jobs < 0:
+            raise ValueError("num_jobs must be nonnegative")
+        if not isfinite(self.arrival_rate) or self.arrival_rate <= 0:
+            raise ValueError("arrival_rate must be finite and positive")
+        if not isfinite(self.deadline_factor) or self.deadline_factor <= 0:
+            raise ValueError("deadline_factor must be finite and positive")
+        low, high = self.duration_range
+        if low < 1 or high < low:
+            raise ValueError("duration_range must satisfy 1 <= minimum <= maximum")
+        if self.burst_size < 1 or self.burst_window < 1:
+            raise ValueError("burst_size and burst_window must be positive")
+
+
+def _steady_arrivals(rng: random.Random, config: WorkloadConfig) -> list[int]:
+    time = 0.0
+    arrivals: list[int] = []
+    for _ in range(config.num_jobs):
+        arrivals.append(int(time))
+        time += rng.expovariate(config.arrival_rate)
+    return arrivals
+
+
+def _make_jobs(
+    arrivals: list[int], rng: random.Random, config: WorkloadConfig, *, mixed: bool = False
+) -> list[Job]:
+    jobs: list[Job] = []
+    for index, arrival in enumerate(sorted(arrivals)):
+        duration = rng.randint(*config.duration_range)
+        if mixed:
+            job_type, weight, factor = rng.choice(
+                (("interactive", 4, 0.5), ("report", 2, 1.0), ("maintenance", 1, 2.0))
+            )
+        else:
+            job_type = rng.choice(("email", "image", "backup"))
+            weight = rng.choice((1, 2, 4))
+            factor = 1.0
+        budget = max(1, ceil(duration * config.deadline_factor * factor * rng.uniform(0.75, 1.25)))
+        jobs.append(Job(f"job-{index:03d}", arrival, duration, arrival + budget, weight, job_type))
+    return jobs
+
+
+def stable_arrivals(seed: int, config: WorkloadConfig = WorkloadConfig()) -> list[Job]:
+    """Constant-rate arrivals, rounded down to integer time units."""
+    rng = random.Random(seed)
+    return _make_jobs(_steady_arrivals(rng, config), rng, config)
+
+
+def bursty_arrivals(seed: int, config: WorkloadConfig = WorkloadConfig()) -> list[Job]:
+    """Batches arriving within short windows, separated by random gaps."""
+    rng = random.Random(seed)
+    time = 0.0
+    arrivals: list[int] = []
+    for offset in range(0, config.num_jobs, config.burst_size):
+        size = min(config.burst_size, config.num_jobs - offset)
+        arrivals.extend(int(time) + rng.randrange(config.burst_window) for _ in range(size))
+        time += rng.expovariate(config.arrival_rate / config.burst_size)
+    return _make_jobs(arrivals, rng, config)
+
+
+def mixed_workload(seed: int, config: WorkloadConfig = WorkloadConfig()) -> list[Job]:
+    """Constant-rate arrivals with tight/high-weight and loose/low-weight jobs."""
+    rng = random.Random(seed)
+    return _make_jobs(_steady_arrivals(rng, config), rng, config, mixed=True)
+
+
+WORKLOADS: dict[str, Callable[[int, WorkloadConfig], list[Job]]] = {
+    "stable": stable_arrivals,
+    "bursty": bursty_arrivals,
+    "mixed": mixed_workload,
+}
diff --git a/tests/test_cli.py b/tests/test_cli.py
new file mode 100644
index 0000000..e3a19b5
--- /dev/null
+++ b/tests/test_cli.py
@@ -0,0 +1,43 @@
+import subprocess
+import sys
+
+
+def test_module_cli_displays_metrics() -> None:
+    result = subprocess.run(
+        [
+            sys.executable,
+            "-m",
+            "adaptivequeue",
+            "--workload",
+            "mixed",
+            "--policy",
+            "edf",
+            "--seed",
+            "7",
+            "--jobs",
+            "12",
+        ],
+        check=True,
+        capture_output=True,
+        text=True,
+    )
+    assert "Workload: mixed | Policy: edf | Seed: 7 | Jobs: 12" in result.stdout
+    for label in (
+        "Primary cost J",
+        "Deadline miss rate",
+        "Average waiting time",
+        "95th-percentile waiting time",
+        "Average flow time",
+    ):
+        assert label in result.stdout
+
+
+def test_invalid_cli_configuration_has_a_useful_error() -> None:
+    result = subprocess.run(
+        [sys.executable, "-m", "adaptivequeue", "--arrival-rate", "0"],
+        capture_output=True,
+        text=True,
+    )
+    assert result.returncode == 2
+    assert "arrival_rate must be finite and positive" in result.stderr
+    assert "Traceback" not in result.stderr
diff --git a/tests/test_metrics.py b/tests/test_metrics.py
new file mode 100644
index 0000000..f44f1dc
--- /dev/null
+++ b/tests/test_metrics.py
@@ -0,0 +1,44 @@
+import pytest
+
+from adaptivequeue import Job, make_policy, simulate
+
+
+def test_job_times_and_metrics_match_hand_calculation() -> None:
+    jobs = [
+        Job("a", 0, 3, 3, 1, "test"),
+        Job("b", 1, 2, 4, 4, "test"),
+        Job("c", 5, 1, 5, 2, "test"),
+    ]
+    result = simulate(jobs, make_policy("fifo", seed=0))
+    assert [(r.start_time, r.completion_time) for r in result.jobs] == [(0, 3), (3, 5), (5, 6)]
+    assert [r.waiting_time for r in result.jobs] == [0, 2, 0]
+    assert [r.flow_time for r in result.jobs] == [3, 4, 1]
+    assert [r.tardiness for r in result.jobs] == [0, 1, 1]
+
+    metrics = result.metrics
+    # Per-job costs: 0.3, 4 * (1 + 0.4), 2 * (1 + 0.1).
+    assert metrics.primary_cost == pytest.approx(2.7)
+    assert metrics.deadline_miss_rate == pytest.approx(2 / 3)
+    assert metrics.average_waiting_time == pytest.approx(2 / 3)
+    assert metrics.p95_waiting_time == pytest.approx(1.8)
+    assert metrics.average_flow_time == pytest.approx(8 / 3)
+
+
+def test_empty_episode_has_zero_metrics_and_never_calls_policy() -> None:
+    def fail_if_called(waiting: object) -> Job:
+        pytest.fail("an empty episode must not request a scheduling decision")
+
+    result = simulate([], fail_if_called)
+    assert result.jobs == ()
+    assert result.metrics.primary_cost == 0
+    assert result.metrics.deadline_miss_rate == 0
+    assert result.metrics.average_waiting_time == 0
+    assert result.metrics.p95_waiting_time == 0
+    assert result.metrics.average_flow_time == 0
+
+
+def test_single_job_finishing_on_deadline_is_not_a_miss() -> None:
+    result = simulate([Job("a", 10, 2, 12, 4, "test")], make_policy("fifo", seed=0))
+    assert result.metrics.deadline_miss_rate == 0
+    assert result.metrics.p95_waiting_time == 0
+    assert result.metrics.primary_cost == pytest.approx(0.8)
diff --git a/tests/test_policies.py b/tests/test_policies.py
new file mode 100644
index 0000000..c66018a
--- /dev/null
+++ b/tests/test_policies.py
@@ -0,0 +1,51 @@
+import pytest
+
+from adaptivequeue import Job, make_policy, simulate
+from adaptivequeue.policies import POLICIES
+
+
+@pytest.mark.parametrize(
+    ("name", "expected"),
+    [
+        ("fifo", ["a", "b", "c", "d"]),
+        ("spt", ["c", "b", "d", "a"]),
+        ("edf", ["c", "d", "a", "b"]),
+        ("priority", ["b", "d", "c", "a"]),
+    ],
+)
+def test_scheduling_order(name: str, expected: list[str]) -> None:
+    jobs = [
+        Job("d", 0, 2, 4, 4, "test"),
+        Job("b", 0, 2, 9, 4, "test"),
+        Job("a", 0, 4, 7, 1, "test"),
+        Job("c", 0, 1, 3, 2, "test"),
+    ]
+    result = simulate(jobs, make_policy(name, seed=1))
+    assert [record.job.job_id for record in result.jobs] == expected
+
+
+@pytest.mark.parametrize("name", POLICIES)
+def test_ties_use_arrival_then_job_id(name: str) -> None:
+    jobs = [
+        Job("b", 1, 2, 9, 2, "test"),
+        Job("a", 1, 2, 9, 2, "test"),
+        Job("z", 0, 2, 9, 2, "test"),
+    ]
+    policy = make_policy(name, seed=1)
+    assert policy(jobs).job_id == "z"
+    assert policy(jobs[:2]).job_id == "a"
+
+
+def test_random_policy_replays_seeded_choices() -> None:
+    jobs = [Job(str(i), 0, 1, 10, 1, "test") for i in range(8)]
+    first = simulate(jobs, make_policy("random", seed=42))
+    replay = simulate(reversed(jobs), make_policy("random", seed=42))
+    other = simulate(jobs, make_policy("random", seed=7))
+    assert first == replay
+    assert [record.job.job_id for record in first.jobs] == ["1", "0", "7", "4", "3", "2", "5", "6"]
+    assert first.jobs != other.jobs
+
+
+def test_unknown_policy_is_rejected() -> None:
+    with pytest.raises(ValueError, match="unknown policy"):
+        make_policy("missing", seed=0)
diff --git a/tests/test_simulation.py b/tests/test_simulation.py
new file mode 100644
index 0000000..3a312a2
--- /dev/null
+++ b/tests/test_simulation.py
@@ -0,0 +1,66 @@
+from typing import Sequence
+
+import pytest
+
+from adaptivequeue import Job, make_policy, simulate
+from adaptivequeue.policies import spt
+
+
+def test_arrivals_do_not_preempt_and_idle_time_is_skipped() -> None:
+    jobs = [
+        Job("late", 20, 1, 25, 1, "test"),
+        Job("boundary", 8, 2, 10, 2, "test"),
+        Job("short", 3, 1, 4, 4, "test"),
+        Job("long", 2, 6, 20, 1, "test"),
+    ]
+    result = simulate(jobs, make_policy("spt", seed=0))
+    assert [(r.job.job_id, r.start_time, r.completion_time) for r in result.jobs] == [
+        ("long", 2, 8),
+        ("short", 8, 9),
+        ("boundary", 9, 11),
+        ("late", 20, 21),
+    ]
+    assert len(result.jobs) == len(jobs)
+    assert all(r.completion_time - r.start_time == r.job.duration for r in result.jobs)
+
+
+def test_policy_sees_only_available_jobs_including_completion_boundary() -> None:
+    jobs = [
+        Job("running", 0, 5, 20, 1, "test"),
+        Job("waiting", 1, 4, 20, 1, "test"),
+        Job("at-completion", 5, 1, 20, 1, "test"),
+        Job("future", 30, 1, 40, 1, "test"),
+    ]
+    observations = []
+
+    def observe(waiting: Sequence[Job]) -> Job:
+        assert isinstance(waiting, tuple)
+        observations.append(tuple(job.job_id for job in waiting))
+        return spt(waiting)
+
+    result = simulate(jobs, observe)
+    assert observations == [("running",), ("waiting", "at-completion"), ("waiting",), ("future",)]
+    assert [r.start_time for r in result.jobs] == [0, 5, 6, 30]
+
+
+def test_duplicate_ids_are_rejected() -> None:
+    job = Job("duplicate", 0, 1, 5, 1, "test")
+    with pytest.raises(ValueError, match="unique"):
+        simulate([job, job], make_policy("fifo", seed=0))
+
+
+def test_policy_cannot_select_a_future_job() -> None:
+    jobs = [Job("available", 0, 2, 10, 1, "test"), Job("future", 3, 1, 10, 1, "test")]
+    with pytest.raises(ValueError, match="waiting queue"):
+        simulate(jobs, lambda waiting: jobs[1])
+
+
+@pytest.mark.parametrize(
+    ("arrival", "duration", "deadline", "weight"),
+    [(-1, 1, 5, 1), (0, 0, 5, 1), (0, 1, -1, 1), (0, 1, 5, 0), (0.5, 1, 5, 1)],
+)
+def test_invalid_job_values_are_rejected(
+    arrival: int, duration: int, deadline: int, weight: int
+) -> None:
+    with pytest.raises(ValueError):
+        Job("invalid", arrival, duration, deadline, weight, "test")
diff --git a/tests/test_workloads.py b/tests/test_workloads.py
new file mode 100644
index 0000000..a19ddd4
--- /dev/null
+++ b/tests/test_workloads.py
@@ -0,0 +1,69 @@
+import random
+from dataclasses import replace
+
+import pytest
+
+from adaptivequeue import WorkloadConfig, make_policy, simulate
+from adaptivequeue.workloads import WORKLOADS
+
+
+@pytest.mark.parametrize("name", WORKLOADS)
+def test_workload_and_episode_are_reproducible_without_global_rng_changes(name: str) -> None:
+    generate = WORKLOADS[name]
+    config = WorkloadConfig()
+    global_state = random.getstate()
+    jobs = generate(42, config)
+    assert jobs == generate(42, config)
+    assert jobs != generate(7, config)
+    assert simulate(jobs, make_policy("random", seed=42)) == simulate(
+        generate(42, config), make_policy("random", seed=42)
+    )
+    assert random.getstate() == global_state
+    assert len(jobs) == len({job.job_id for job in jobs}) == 60
+    assert all(1 <= job.duration <= 6 and job.priority_weight in (1, 2, 4) for job in jobs)
+    assert all(job.deadline > job.arrival_time and job.job_type for job in jobs)
+
+
+@pytest.mark.parametrize("name", WORKLOADS)
+def test_workload_configuration_controls_size_load_and_deadlines(name: str) -> None:
+    generate = WORKLOADS[name]
+    config = WorkloadConfig(num_jobs=13, duration_range=(2, 2), burst_size=4)
+    jobs = generate(42, config)
+    faster = generate(42, replace(config, arrival_rate=1.0))
+    looser = generate(42, replace(config, deadline_factor=8.0))
+    assert len(jobs) == 13
+    assert all(job.duration == 2 for job in jobs)
+    assert max(job.arrival_time for job in faster) < max(job.arrival_time for job in jobs)
+    assert [job.arrival_time for job in jobs] == [job.arrival_time for job in looser]
+    assert all(a.deadline < b.deadline for a, b in zip(jobs, looser))
+    assert generate(42, WorkloadConfig(num_jobs=0)) == []
+
+
+def test_burst_windows_and_mixed_job_classes() -> None:
+    config = WorkloadConfig(num_jobs=6, burst_size=6, burst_window=3)
+    burst = WORKLOADS["bursty"](42, config)
+    assert all(0 <= job.arrival_time < 3 for job in burst)
+    mixed = WORKLOADS["mixed"](42, WorkloadConfig())
+    assert {(job.job_type, job.priority_weight) for job in mixed} == {
+        ("interactive", 4),
+        ("report", 2),
+        ("maintenance", 1),
+    }
+
+
+@pytest.mark.parametrize(
+    "changes",
+    [
+        {"num_jobs": -1},
+        {"arrival_rate": 0},
+        {"arrival_rate": float("nan")},
+        {"deadline_factor": -1},
+        {"deadline_factor": float("inf")},
+        {"duration_range": (3, 2)},
+        {"burst_size": 0},
+        {"burst_window": 0},
+    ],
+)
+def test_invalid_configuration_is_rejected(changes: dict) -> None:
+    with pytest.raises(ValueError):
+        WorkloadConfig(**changes)

From e505b0a4d53ed3a43ca855dc357dfacb2c502615 Mon Sep 17 00:00:00 2001
From: TohruSu <238259854+TohruSu@users.noreply.github.com>
Date: Thu, 15 Oct 2026 11:27:53 +0000
Subject: [PATCH 2/2] Pass current simulation time to scheduling policies

---
 docs/simulation.md              |  6 ++++--
 src/adaptivequeue/policies.py   | 18 +++++++++++-------
 src/adaptivequeue/simulation.py |  2 +-
 tests/test_metrics.py           |  2 +-
 tests/test_policies.py          |  4 ++--
 tests/test_simulation.py        | 17 +++++++++++------
 6 files changed, 30 insertions(+), 19 deletions(-)

diff --git a/docs/simulation.md b/docs/simulation.md
index efb31a8..9c424ff 100644
--- a/docs/simulation.md
+++ b/docs/simulation.md
@@ -10,8 +10,9 @@ can be reused across policies.
 
 One worker runs one job to completion without preemption. Whenever it is idle,
 all jobs with arrival time at or before the current time enter the waiting queue.
-The policy receives only an immutable snapshot of that queue and must select a
-waiting job. It cannot observe future arrivals. Jobs arriving during execution
+The engine calls `policy(waiting_jobs, current_time)` with an immutable snapshot
+of that queue and the current integer simulation time. The policy must select a
+waiting job and cannot observe future arrivals. Jobs arriving during execution
 wait until completion; arrivals exactly at completion participate in the next
 decision. The engine skips to completion or the next arrival rather than stepping
 through ticks where no scheduling decision is possible. It runs until every job
@@ -27,6 +28,7 @@ finishes, including late jobs, without deliberately idling when work is availabl
 | `priority` | Highest priority weight |
 | `random` | Uniformly random waiting job, using a local seeded RNG |
 
+These baselines accept `current_time` but do not use it in their selection rules.
 The first four break ties by earliest arrival, then lexicographically smallest
 job ID. Random's queue is populated in arrival/ID order, making seeded results
 independent of input ordering. Recreate a Random policy with the same seed for
diff --git a/src/adaptivequeue/policies.py b/src/adaptivequeue/policies.py
index 343db64..eb83e07 100644
--- a/src/adaptivequeue/policies.py
+++ b/src/adaptivequeue/policies.py
@@ -1,26 +1,26 @@
-"""Policies receive only the jobs currently waiting for the worker."""
+"""Policies receive the waiting jobs and current simulation time."""
 
 import random
 from typing import Callable, Sequence
 
 from adaptivequeue.models import Job
 
-Policy = Callable[[Sequence[Job]], Job]
+Policy = Callable[[Sequence[Job], int], Job]
 
 
-def fifo(waiting: Sequence[Job]) -> Job:
+def fifo(waiting: Sequence[Job], current_time: int) -> Job:
     return min(waiting, key=lambda job: (job.arrival_time, job.job_id))
 
 
-def spt(waiting: Sequence[Job]) -> Job:
+def spt(waiting: Sequence[Job], current_time: int) -> Job:
     return min(waiting, key=lambda job: (job.duration, job.arrival_time, job.job_id))
 
 
-def edf(waiting: Sequence[Job]) -> Job:
+def edf(waiting: Sequence[Job], current_time: int) -> Job:
     return min(waiting, key=lambda job: (job.deadline, job.arrival_time, job.job_id))
 
 
-def priority(waiting: Sequence[Job]) -> Job:
+def priority(waiting: Sequence[Job], current_time: int) -> Job:
     return min(waiting, key=lambda job: (-job.priority_weight, job.arrival_time, job.job_id))
 
 
@@ -32,7 +32,11 @@ def make_policy(name: str, *, seed: int) -> Policy:
     """Create a policy; recreate it with the same seed to replay a random run."""
     if name == "random":
         rng = random.Random(seed)
-        return rng.choice
+
+        def random_policy(waiting: Sequence[Job], current_time: int) -> Job:
+            return rng.choice(waiting)
+
+        return random_policy
     try:
         return POLICIES[name]
     except KeyError:
diff --git a/src/adaptivequeue/simulation.py b/src/adaptivequeue/simulation.py
index 7ff5e3a..f373489 100644
--- a/src/adaptivequeue/simulation.py
+++ b/src/adaptivequeue/simulation.py
@@ -32,7 +32,7 @@ def simulate(jobs: Iterable[Job], policy: Policy) -> SimulationResult:
             waiting.append(arrivals[next_arrival])
             next_arrival += 1
 
-        selected = policy(tuple(waiting))
+        selected = policy(tuple(waiting), time)
         if selected not in waiting:
             raise ValueError("policy must select a job from the waiting queue")
         waiting.remove(selected)
diff --git a/tests/test_metrics.py b/tests/test_metrics.py
index f44f1dc..8601c81 100644
--- a/tests/test_metrics.py
+++ b/tests/test_metrics.py
@@ -25,7 +25,7 @@ def test_job_times_and_metrics_match_hand_calculation() -> None:
 
 
 def test_empty_episode_has_zero_metrics_and_never_calls_policy() -> None:
-    def fail_if_called(waiting: object) -> Job:
+    def fail_if_called(waiting: object, current_time: int) -> Job:
         pytest.fail("an empty episode must not request a scheduling decision")
 
     result = simulate([], fail_if_called)
diff --git a/tests/test_policies.py b/tests/test_policies.py
index c66018a..d31d2ec 100644
--- a/tests/test_policies.py
+++ b/tests/test_policies.py
@@ -32,8 +32,8 @@ def test_ties_use_arrival_then_job_id(name: str) -> None:
         Job("z", 0, 2, 9, 2, "test"),
     ]
     policy = make_policy(name, seed=1)
-    assert policy(jobs).job_id == "z"
-    assert policy(jobs[:2]).job_id == "a"
+    assert policy(jobs, 1).job_id == "z"
+    assert policy(jobs[:2], 1).job_id == "a"
 
 
 def test_random_policy_replays_seeded_choices() -> None:
diff --git a/tests/test_simulation.py b/tests/test_simulation.py
index 3a312a2..d277f18 100644
--- a/tests/test_simulation.py
+++ b/tests/test_simulation.py
@@ -24,7 +24,7 @@ def test_arrivals_do_not_preempt_and_idle_time_is_skipped() -> None:
     assert all(r.completion_time - r.start_time == r.job.duration for r in result.jobs)
 
 
-def test_policy_sees_only_available_jobs_including_completion_boundary() -> None:
+def test_policy_sees_available_jobs_and_current_time_at_each_decision() -> None:
     jobs = [
         Job("running", 0, 5, 20, 1, "test"),
         Job("waiting", 1, 4, 20, 1, "test"),
@@ -33,13 +33,18 @@ def test_policy_sees_only_available_jobs_including_completion_boundary() -> None
     ]
     observations = []
 
-    def observe(waiting: Sequence[Job]) -> Job:
+    def observe(waiting: Sequence[Job], current_time: int) -> Job:
         assert isinstance(waiting, tuple)
-        observations.append(tuple(job.job_id for job in waiting))
-        return spt(waiting)
+        observations.append((current_time, tuple(job.job_id for job in waiting)))
+        return spt(waiting, current_time)
 
     result = simulate(jobs, observe)
-    assert observations == [("running",), ("waiting", "at-completion"), ("waiting",), ("future",)]
+    assert observations == [
+        (0, ("running",)),
+        (5, ("waiting", "at-completion")),
+        (6, ("waiting",)),
+        (30, ("future",)),
+    ]
     assert [r.start_time for r in result.jobs] == [0, 5, 6, 30]
 
 
@@ -52,7 +57,7 @@ def test_duplicate_ids_are_rejected() -> None:
 def test_policy_cannot_select_a_future_job() -> None:
     jobs = [Job("available", 0, 2, 10, 1, "test"), Job("future", 3, 1, 10, 1, "test")]
     with pytest.raises(ValueError, match="waiting queue"):
-        simulate(jobs, lambda waiting: jobs[1])
+        simulate(jobs, lambda waiting, current_time: jobs[1])
 
 
 @pytest.mark.parametrize(