Bug report (python repo "scheduler"):
The scheduler hands us more work than our resource pool can support.

With a pool of 3 "cpu" units and two tasks that each request 2, a single call to
next_batch comes back with both of them. Starting both then blows up with a
ResourceExhausted from deep inside the pool.

Each task on its own fits, so the pool's accounting looks right to us -- it's
the batch that seems wrong. Please fix it so a batch is always startable.

Repository files (containing the defects):
README.md: # flow

A small workflow engine: directed graphs of tasks, run state, and a scheduler
that decides what may run next.

```python
from flow import Task, TaskGraph, ResourcePool, Scheduler

graph = TaskGraph([
    Task("fetch"),
    Task("build", depends_on=frozenset({"fetch"})),
    Task("test",  depends_on=frozenset({"build"})),
])
scheduler = Scheduler(graph, ResourcePool({}))
state = scheduler.new_state()

for task_id in scheduler.next_batch(state):
    scheduler.start(task_id, state)
    scheduler.finish(task_id, state, result=True)
```

## Layout

| module | role |
|---|---|
| `task.py` | `Task`, `ResourceRequest` — what a unit of work declares |
| `graph.py` | `TaskGraph` — dependency structure, traversal, topological order |
| `state.py` | `RunState` — per-task status, the legal transitions between them |
| `scheduler.py` | eligibility, batching, and the start/finish/fail transitions |
| `resources.py` | `ResourcePool` — named capacities held for the duration of a task |
| `retry.py` | `RetryPolicy` — attempt counting and deterministic backoff |
| `calendar.py` | `Calendar`, `Window` — when tasks are permitted to start |
| `expressions.py` | the condition mini-language used to gate tasks |
| `validation.py` | static checks run before a graph executes |
| `serialize.py` | lossless persistence of a run in progress |
| `metrics.py` | counters and an event timeline |
| `clock.py` | injectable time source, so runs are reproducible |

## Design notes

**Determinism.** Anything that could vary between runs is pinned. Ties in
priority are broken by task id, topological order sorts its ready set, and
retry backoff carries no jitter. Two runs of the same graph make the same
decisions in the same order.

**The scheduler does not execute.** It reports what is eligible; the caller
runs the work and reports back. Admission rules are therefore testable without
running anything.

**Half-open intervals.** A calendar window is `[start, end)` throughout, so
adjacent windows neither overlap nor leave a gap.

**Failure propagates.** A task that will never succeed skips everything
downstream of it, because those tasks depend on a result that will not exist.

flow/__init__.py: """A small workflow engine: graphs of tasks, run state, and a scheduler."""
from .calendar import Calendar, Window
from .clock import Clock
from .errors import (CyclicGraph, DuplicateTask, ExpressionError, FlowError,
                     GraphError, IllegalTransition, ResourceExhausted,
                     SerializationError, UnknownResource, UnknownTask)
from .expressions import evaluate, names_in, tokenize
from .graph import TaskGraph
from .metrics import Metrics
from .resources import ResourcePool
from .retry import NO_RETRY, RetryPolicy
from .scheduler import Scheduler
from .serialize import dump_state, load_state, round_trip
from .state import RunState
from .task import ResourceRequest, Task
from .validation import validate_all

__all__ = [
    "Calendar", "Window", "Clock", "TaskGraph", "Metrics", "ResourcePool",
    "RetryPolicy", "NO_RETRY", "Scheduler", "RunState", "Task",
    "ResourceRequest", "validate_all", "evaluate", "names_in", "tokenize",
    "dump_state", "load_state", "round_trip",
    "FlowError", "GraphError", "CyclicGraph", "DuplicateTask", "UnknownTask",
    "IllegalTransition", "ResourceExhausted", "UnknownResource",
    "ExpressionError", "SerializationError",
]

flow/calendar.py: """Scheduling windows.

A window is a half-open interval [start, end): a task may begin exactly at
`start`, and may not begin at `end`. Half-open intervals are used throughout so
that back-to-back windows neither overlap nor leave a gap.
"""
from dataclasses import dataclass
from typing import List, Optional, Sequence


@dataclass(frozen=True)
class Window:
    start: float
    end: float

    def __post_init__(self):
        if self.end <= self.start:
            raise ValueError("window end must be after its start")

    def contains(self, when: float) -> bool:
        return self.start <= when < self.end

    def __str__(self) -> str:
        return f"[{self.start}, {self.end})"


class Calendar:
    """The times at which tasks are permitted to start.

    An empty calendar permits everything; that is the common case and avoids
    forcing every caller to describe an always-open schedule.
    """

    def __init__(self, windows: Optional[Sequence[Window]] = None):
        self.windows: List[Window] = sorted(windows or (),
                                            key=lambda w: (w.start, w.end))

    def is_open(self, when: float) -> bool:
        if not self.windows:
            return True
        return any(window.contains(when) for window in self.windows)

    def next_open(self, when: float) -> Optional[float]:
        """The earliest permitted time at or after `when`."""
        if not self.windows:
            return when
        if self.is_open(when):
            return when
        for window in self.windows:
            if window.start >= when:
                return window.start
        return None

    def closes_at(self, when: float) -> Optional[float]:
        for window in self.windows:
            if window.contains(when):
                return window.end
        return None

flow/clock.py: """Time source.

Everything that needs the current time takes a Clock, so a run can be replayed
deterministically in tests instead of depending on the wall clock.
"""


class Clock:
    """Monotonic seconds since the run started."""

    def __init__(self, start: float = 0.0):
        self._now = float(start)

    def now(self) -> float:
        return self._now

    def advance(self, seconds: float) -> float:
        if seconds < 0:
            raise ValueError("cannot move a clock backwards")
        self._now += float(seconds)
        return self._now

flow/errors.py: """Exception hierarchy for the workflow engine."""


class FlowError(Exception):
    """Base class for everything this package raises."""


class GraphError(FlowError):
    """The task graph is malformed."""


class CyclicGraph(GraphError):
    def __init__(self, cycle):
        super().__init__("cycle: " + " -> ".join(cycle))
        self.cycle = list(cycle)


class UnknownTask(FlowError):
    def __init__(self, task_id: str):
        super().__init__(f"unknown task: {task_id}")
        self.task_id = task_id


class DuplicateTask(GraphError):
    def __init__(self, task_id: str):
        super().__init__(f"task already defined: {task_id}")
        self.task_id = task_id


class IllegalTransition(FlowError):
    def __init__(self, task_id: str, old: str, new: str):
        super().__init__(f"{task_id}: cannot move from {old} to {new}")
        self.task_id = task_id
        self.old = old
        self.new = new


class ResourceExhausted(FlowError):
    def __init__(self, name: str, want: int, free: int):
        super().__init__(f"resource {name!r}: wanted {want}, {free} free")
        self.name = name
        self.want = want
        self.free = free


class UnknownResource(FlowError):
    def __init__(self, name: str):
        super().__init__(f"unknown resource: {name}")
        self.name = name


class ExpressionError(FlowError):
    """A condition expression could not be parsed or evaluated."""


class SerializationError(FlowError):
    """A persisted run could not be restored."""

flow/expressions.py: """A very small condition language for gating tasks.

Grammar, loosest binding first:

    expr     := or_expr
    or_expr  := and_expr ( 'or' and_expr )*
    and_expr := unary ( 'and' unary )*
    unary    := 'not' unary | atom
    atom     := '(' expr ')' | NAME

A NAME resolves against a mapping of facts, so `build and not skip_tests` is
true when `build` is truthy and `skip_tests` is not. `or` binds more loosely
than `and`, as it does in Python.
"""
import re
from typing import List, Mapping, Optional

from .errors import ExpressionError

_TOKEN = re.compile(r"\s*(\(|\)|[A-Za-z_][A-Za-z0-9_.]*)")
_KEYWORDS = ("and", "or", "not")


def tokenize(text: str) -> List[str]:
    tokens: List[str] = []
    pos = 0
    while pos < len(text):
        if text[pos].isspace():
            pos += 1
            continue
        match = _TOKEN.match(text, pos)
        if not match:
            raise ExpressionError(f"unexpected character {text[pos]!r} at {pos}")
        tokens.append(match.group(1))
        pos = match.end()
    return tokens


def names_in(text: str) -> List[str]:
    """Every fact name an expression refers to."""
    return sorted({t for t in tokenize(text)
                   if t not in _KEYWORDS and t not in ("(", ")")})


class _Parser:
    def __init__(self, tokens: List[str], facts: Mapping[str, object]):
        self.tokens = tokens
        self.pos = 0
        self.facts = facts

    def peek(self) -> Optional[str]:
        return self.tokens[self.pos] if self.pos < len(self.tokens) else None

    def take(self) -> str:
        token = self.peek()
        if token is None:
            raise ExpressionError("unexpected end of expression")
        self.pos += 1
        return token

    def parse(self) -> bool:
        value = self.or_expr()
        if self.peek() is not None:
            raise ExpressionError(f"trailing input at {self.peek()!r}")
        return value

    def or_expr(self) -> bool:
        value = self.and_expr()
        while self.peek() == "or":
            self.take()
            right = self.and_expr()
            value = value or right
        return value

    def and_expr(self) -> bool:
        value = self.unary()
        while self.peek() == "and":
            self.take()
            right = self.unary()
            value = value and right
        return value

    def unary(self) -> bool:
        if self.peek() == "not":
            self.take()
            return not self.unary()
        return self.atom()

    def atom(self) -> bool:
        token = self.take()
        if token == "(":
            value = self.or_expr()
            if self.take() != ")":
                raise ExpressionError("missing closing parenthesis")
            return value
        if token in _KEYWORDS or token == ")":
            raise ExpressionError(f"unexpected token {token!r}")
        return bool(self.facts.get(token, False))


def evaluate(expression: Optional[str], facts: Mapping[str, object]) -> bool:
    """Evaluate a condition. An absent or blank condition is always true."""
    if expression is None or not expression.strip():
        return True
    tokens = tokenize(expression)
    if not tokens:
        return True
    return _Parser(tokens, facts).parse()

flow/graph.py: """The task graph: dependency structure and traversal."""
from typing import Dict, Iterable, Iterator, List, Optional, Set

from .errors import CyclicGraph, DuplicateTask, UnknownTask
from .task import Task


class TaskGraph:
    """A directed acyclic graph of tasks, keyed by task id.

    Edges point from a dependency to the task that needs it: if B depends on A
    then A is *upstream* of B, and B is a *dependent* of A.
    """

    def __init__(self, tasks: Optional[Iterable[Task]] = None):
        self._tasks: Dict[str, Task] = {}
        self._dependents: Dict[str, Set[str]] = {}
        for task in tasks or ():
            self.add(task)

    # -- construction ----------------------------------------------------

    def add(self, task: Task) -> Task:
        if task.task_id in self._tasks:
            raise DuplicateTask(task.task_id)
        self._tasks[task.task_id] = task
        self._dependents.setdefault(task.task_id, set())
        for upstream in task.depends_on:
            self._dependents.setdefault(upstream, set()).add(task.task_id)
        return task

    def validate(self) -> None:
        """Check every dependency resolves and the graph is acyclic."""
        for task in self._tasks.values():
            for upstream in task.depends_on:
                if upstream not in self._tasks:
                    raise UnknownTask(upstream)
        self.topological_order()

    # -- access ----------------------------------------------------------

    def __len__(self) -> int:
        return len(self._tasks)

    def __contains__(self, task_id: object) -> bool:
        return task_id in self._tasks

    def __iter__(self) -> Iterator[Task]:
        return iter(self._tasks.values())

    def get(self, task_id: str) -> Task:
        if task_id not in self._tasks:
            raise UnknownTask(task_id)
        return self._tasks[task_id]

    def ids(self) -> List[str]:
        return sorted(self._tasks)

    def upstream_of(self, task_id: str) -> Set[str]:
        """Direct dependencies of `task_id`."""
        return set(self.get(task_id).depends_on)

    def dependents_of(self, task_id: str) -> Set[str]:
        """Tasks that directly depend on `task_id`."""
        if task_id not in self._tasks:
            raise UnknownTask(task_id)
        return set(self._dependents.get(task_id, ()))

    def roots(self) -> List[str]:
        """Tasks with no dependencies, in sorted order."""
        return sorted(t.task_id for t in self._tasks.values() if not t.depends_on)

    def leaves(self) -> List[str]:
        return sorted(tid for tid in self._tasks
                      if not self._dependents.get(tid))

    # -- traversal -------------------------------------------------------

    def descendants_of(self, task_id: str) -> Set[str]:
        """Everything reachable downstream of `task_id`, excluding itself."""
        seen: Set[str] = set()
        stack = list(self.dependents_of(task_id))
        while stack:
            current = stack.pop()
            if current in seen:
                continue
            seen.add(current)
            stack.extend(self._dependents.get(current, ()))
        return seen

    def ancestors_of(self, task_id: str) -> Set[str]:
        """Everything `task_id` transitively depends on."""
        seen: Set[str] = set()
        stack = list(self.upstream_of(task_id))
        while stack:
            current = stack.pop()
            if current in seen:
                continue
            seen.add(current)
            stack.extend(self.get(current).depends_on)
        return seen

    def topological_order(self) -> List[str]:
        """A dependency-respecting order, ties broken by id for determinism."""
        indegree = {tid: len(self._tasks[tid].depends_on) for tid in self._tasks}
        ready = sorted(tid for tid, n in indegree.items() if n == 0)
        order: List[str] = []
        while ready:
            current = ready.pop(0)
            order.append(current)
            for downstream in sorted(self._dependents.get(current, ())):
                indegree[downstream] -= 1
                if indegree[downstream] == 0:
                    ready.append(downstream)
            ready.sort()
        if len(order) != len(self._tasks):
            raise CyclicGraph(sorted(set(self._tasks) - set(order)))
        return order

flow/metrics.py: """Counters and an event timeline collected over a run."""
from typing import Dict, List, Tuple


class Metrics:
    def __init__(self) -> None:
        self.counters: Dict[str, int] = {}
        self.timeline: List[Tuple[float, str, str]] = []

    def bump(self, name: str, by: int = 1) -> int:
        self.counters[name] = self.counters.get(name, 0) + by
        return self.counters[name]

    def record(self, when: float, task_id: str, event: str) -> None:
        self.timeline.append((when, task_id, event))
        self.bump(event)

    def events_for(self, task_id: str) -> List[str]:
        return [event for _, tid, event in self.timeline if tid == task_id]

    def count(self, name: str) -> int:
        return self.counters.get(name, 0)

    def summary(self) -> Dict[str, int]:
        return dict(sorted(self.counters.items()))

flow/resources.py: """Named resource pools with fixed capacity."""
from typing import Dict, Iterable, List, Mapping, Tuple

from .errors import ResourceExhausted, UnknownResource
from .task import ResourceRequest


class ResourcePool:
    """Tracks how much of each named resource is currently held.

    Holdings are recorded per task so that releasing is exact: a task gives
    back precisely what it took, which is what keeps a failed attempt from
    leaking capacity for the rest of the run.
    """

    def __init__(self, capacity: Mapping[str, int]):
        for name, total in capacity.items():
            if total < 0:
                raise ValueError(f"capacity for {name!r} must not be negative")
        self.capacity: Dict[str, int] = dict(capacity)
        self._held: Dict[str, Dict[str, int]] = {}

    def free(self, name: str) -> int:
        if name not in self.capacity:
            raise UnknownResource(name)
        used = sum(held.get(name, 0) for held in self._held.values())
        return self.capacity[name] - used

    def can_admit(self, requests: Iterable[ResourceRequest]) -> bool:
        for request in requests:
            if request.name not in self.capacity:
                raise UnknownResource(request.name)
            if self.free(request.name) < request.amount:
                return False
        return True

    def acquire(self, task_id: str, requests: Iterable[ResourceRequest]) -> None:
        """Take every request at once, or nothing at all."""
        requests = list(requests)
        for request in requests:
            if request.name not in self.capacity:
                raise UnknownResource(request.name)
            if self.free(request.name) < request.amount:
                raise ResourceExhausted(request.name, request.amount,
                                        self.free(request.name))
        holding = self._held.setdefault(task_id, {})
        for request in requests:
            holding[request.name] = holding.get(request.name, 0) + request.amount

    def release(self, task_id: str) -> Dict[str, int]:
        """Give back everything `task_id` holds. Safe to call twice."""
        return self._held.pop(task_id, {})

    def held_by(self, task_id: str) -> Dict[str, int]:
        return dict(self._held.get(task_id, {}))

    def in_use(self) -> Dict[str, int]:
        out: Dict[str, int] = {name: 0 for name in self.capacity}
        for holding in self._held.values():
            for name, amount in holding.items():
                out[name] = out.get(name, 0) + amount
        return out

flow/retry.py: """Retry policies.

Backoff is deterministic on purpose: a scheduler test that depended on jitter
would be flaky, and the engine's own behaviour must be reproducible.
"""
from dataclasses import dataclass


@dataclass(frozen=True)
class RetryPolicy:
    """How many times to retry, and how long to wait between attempts.

    `max_attempts` counts the *total* number of attempts, so the default of 1
    means a task runs once and is not retried.
    """
    max_attempts: int = 1
    base_delay: float = 1.0
    multiplier: float = 2.0
    max_delay: float = 300.0

    def __post_init__(self):
        if self.max_attempts < 1:
            raise ValueError("max_attempts must be at least 1")
        if self.base_delay < 0:
            raise ValueError("base_delay must not be negative")
        if self.multiplier < 1.0:
            raise ValueError("multiplier must be at least 1.0")

    def should_retry(self, attempts_made: int) -> bool:
        """True when another attempt is still allowed."""
        return attempts_made < self.max_attempts

    def delay_for(self, attempts_made: int) -> float:
        """Seconds to wait before attempt number `attempts_made` + 1.

        The first retry waits `base_delay`; each subsequent one multiplies by
        `multiplier`, clamped at `max_delay`.
        """
        if attempts_made < 1:
            return 0.0
        delay = self.base_delay * (self.multiplier ** (attempts_made - 1))
        return min(delay, self.max_delay)


NO_RETRY = RetryPolicy(max_attempts=1)

flow/scheduler.py: """The scheduler: decides what may run, and drives a run to completion."""
from typing import Dict, List, Mapping, Optional, Sequence

from . import state as st
from .calendar import Calendar
from .clock import Clock
from .expressions import evaluate
from .graph import TaskGraph
from .metrics import Metrics
from .resources import ResourcePool
from .task import Task


class Scheduler:
    """Turns a graph plus a run state into a sequence of dispatch decisions.

    The scheduler never executes anything itself. `next_batch` reports what is
    eligible right now, and the caller reports back what happened. That split
    keeps the admission rules testable without running any work.
    """

    def __init__(self, graph: TaskGraph, pool: ResourcePool,
                 calendar: Optional[Calendar] = None,
                 clock: Optional[Clock] = None,
                 metrics: Optional[Metrics] = None,
                 max_parallel: int = 0):
        self.graph = graph
        self.pool = pool
        self.calendar = calendar or Calendar()
        self.clock = clock or Clock()
        self.metrics = metrics or Metrics()
        self.max_parallel = max_parallel

    # -- eligibility -----------------------------------------------------

    def upstream_satisfied(self, task: Task, state: st.RunState) -> bool:
        """True when every dependency has finished successfully.

        A dependency that was skipped does not satisfy anything downstream:
        its result was never produced, so anything needing it is skipped too.
        """
        for upstream in task.depends_on:
            if state.get(upstream) != st.SUCCEEDED:
                return False
        return True

    def upstream_settled(self, task: Task, state: st.RunState) -> bool:
        """True once no dependency can change state again."""
        return all(state.is_terminal(u) for u in task.depends_on)

    def condition_holds(self, task: Task, state: st.RunState) -> bool:
        """Evaluate the task's condition against upstream results."""
        if not task.condition:
            return True
        facts = {tid: state.results.get(tid, False)
                 for tid in self.graph.ancestors_of(task.task_id)
                 if state.get(tid) == st.SUCCEEDED}
        return evaluate(task.condition, facts)

    def running_count(self, state: st.RunState) -> int:
        return len(state.in_status(st.RUNNING))

    def eligible(self, state: st.RunState) -> List[str]:
        """Every task that could be dispatched now, in dispatch order."""
        when = self.clock.now()
        if not self.calendar.is_open(when):
            return []

        candidates: List[Task] = []
        for task_id in self.graph.ids():
            # RETRYING counts as a candidate: its backoff is enforced below by
            # ready_at, and leaving it out would mean a retry never fires.
            if state.get(task_id) not in (st.PENDING, st.READY, st.RETRYING):
                continue
            task = self.graph.get(task_id)
            if not self.upstream_satisfied(task, state):
                continue
            if not self.condition_holds(task, state):
                continue
            earliest = state.ready_at.get(task_id)
            if earliest is not None and when < earliest:
                continue
            candidates.append(task)

        candidates.sort(key=lambda t: t.sort_key())
        return [t.task_id for t in candidates]

    def next_batch(self, state: st.RunState) -> List[str]:
        """The eligible tasks that also fit within resources and parallelism.

        Resource checks are cumulative across the batch: two tasks that each
        fit on their own must not both be admitted when together they exceed
        the pool.
        """
        chosen: List[str] = []
        in_flight = self.running_count(state)

        for task_id in self.eligible(state):
            if self.max_parallel and in_flight + len(chosen) >= self.max_parallel:
                break
            task = self.graph.get(task_id)
            if not self.pool.can_admit(task.resources):
                continue
            chosen.append(task_id)
        return chosen

    # -- transitions -----------------------------------------------------

    def start(self, task_id: str, state: st.RunState) -> None:
        task = self.graph.get(task_id)
        if state.get(task_id) == st.PENDING:
            state.set(task_id, st.READY)
        elif state.get(task_id) == st.RETRYING:
            state.set(task_id, st.READY)
        self.pool.acquire(task_id, task.resources)
        state.set(task_id, st.RUNNING)
        state.record_attempt(task_id)
        self.metrics.record(self.clock.now(), task_id, "started")

    def finish(self, task_id: str, state: st.RunState, result: object = None) -> None:
        self.pool.release(task_id)
        state.succeed(task_id, result)
        self.metrics.record(self.clock.now(), task_id, "succeeded")

    def fail(self, task_id: str, state: st.RunState) -> str:
        """Record a failed attempt, retrying it if the policy still allows.

        The pool is released on every failure, not only the last one: a task
        that is going to be retried must not keep holding capacity while it
        waits out its backoff.
        """
        self.pool.release(task_id)
        task = self.graph.get(task_id)
        attempts = state.attempts.get(task_id, 0)
        if task.retry.should_retry(attempts):
            state.set(task_id, st.RETRYING)
            delay = task.retry.delay_for(attempts)
            state.ready_at[task_id] = self.clock.now() + delay
            self.metrics.record(self.clock.now(), task_id, "retrying")
            return st.RETRYING
        state.set(task_id, st.FAILED)
        self.metrics.record(self.clock.now(), task_id, "failed")
        self.cascade_skip(task_id, state)
        return st.FAILED

    def cascade_skip(self, task_id: str, state: st.RunState) -> List[str]:
        """Skip everything downstream of a task that will never succeed."""
        skipped: List[str] = []
        for downstream in sorted(self.graph.descendants_of(task_id)):
            if state.get(downstream) in (st.PENDING, st.READY):
                state.set(downstream, st.SKIPPED)
                self.metrics.record(self.clock.now(), downstream, "skipped")
                skipped.append(downstream)
        return skipped

    def settle_unreachable(self, state: st.RunState) -> List[str]:
        """Skip tasks whose dependencies are settled but not all successful."""
        skipped: List[str] = []
        for task_id in state.in_status(st.PENDING, st.READY):
            task = self.graph.get(task_id)
            if not self.upstream_settled(task, state):
                continue
            if not self.upstream_satisfied(task, state):
                state.set(task_id, st.SKIPPED)
                skipped.append(task_id)
            elif not self.condition_holds(task, state):
                state.set(task_id, st.SKIPPED)
                skipped.append(task_id)
        return skipped

    def new_state(self) -> st.RunState:
        return st.RunState(self.graph.ids())

flow/serialize.py: """Persist and restore a run in progress.

A round trip must be lossless: a run reloaded from disk has to make exactly
the decisions it would have made had it never stopped. Anything the scheduler
reads therefore has to be written.
"""
from typing import Any, Dict

from .errors import SerializationError
from .state import RunState

FORMAT_VERSION = 2


def dump_state(state: RunState) -> Dict[str, Any]:
    return {
        "version": FORMAT_VERSION,
        "status": dict(state.status),
        "attempts": dict(state.attempts),
        "results": dict(state.results),
        "ready_at": dict(state.ready_at),
    }


def load_state(payload: Dict[str, Any]) -> RunState:
    if not isinstance(payload, dict):
        raise SerializationError("payload must be a mapping")
    version = payload.get("version")
    if version != FORMAT_VERSION:
        raise SerializationError(
            f"unsupported format version {version!r}, expected {FORMAT_VERSION}")
    status = payload.get("status")
    if not isinstance(status, dict):
        raise SerializationError("missing status map")

    state = RunState(status)
    state.status = dict(status)
    state.attempts = dict(payload.get("attempts") or {})
    state.results = dict(payload.get("results") or {})
    state.ready_at = dict(payload.get("ready_at") or {})
    for task_id in state.status:
        state.attempts.setdefault(task_id, 0)
    return state


def round_trip(state: RunState) -> RunState:
    return load_state(dump_state(state))

flow/state.py: """Run state: the status of every task in one execution of a graph."""
from typing import Dict, Iterable, List, Optional, Set

from .errors import IllegalTransition, UnknownTask

PENDING = "pending"
READY = "ready"
RUNNING = "running"
SUCCEEDED = "succeeded"
FAILED = "failed"
RETRYING = "retrying"
SKIPPED = "skipped"

TERMINAL = frozenset({SUCCEEDED, FAILED, SKIPPED})

# Which moves the engine permits. A task that has finished stays finished:
# re-running one would silently invalidate everything computed downstream of it.
ALLOWED: Dict[str, Set[str]] = {
    PENDING: {READY, SKIPPED},
    READY: {RUNNING, SKIPPED, PENDING},
    RUNNING: {SUCCEEDED, FAILED, RETRYING},
    RETRYING: {READY, FAILED},
    SUCCEEDED: set(),
    FAILED: set(),
    SKIPPED: set(),
}


class RunState:
    """Mutable status of each task, plus attempt counts and results."""

    def __init__(self, task_ids: Iterable[str]):
        self.status: Dict[str, str] = {tid: PENDING for tid in task_ids}
        self.attempts: Dict[str, int] = {tid: 0 for tid in self.status}
        self.results: Dict[str, object] = {}
        self.ready_at: Dict[str, float] = {}

    def _check(self, task_id: str) -> str:
        if task_id not in self.status:
            raise UnknownTask(task_id)
        return self.status[task_id]

    def get(self, task_id: str) -> str:
        return self._check(task_id)

    def set(self, task_id: str, new: str) -> str:
        old = self._check(task_id)
        if new not in ALLOWED.get(old, set()):
            raise IllegalTransition(task_id, old, new)
        self.status[task_id] = new
        return new

    def record_attempt(self, task_id: str) -> int:
        self._check(task_id)
        self.attempts[task_id] = self.attempts.get(task_id, 0) + 1
        return self.attempts[task_id]

    def succeed(self, task_id: str, result: object = None) -> None:
        self.set(task_id, SUCCEEDED)
        self.results[task_id] = result

    def is_terminal(self, task_id: str) -> bool:
        return self._check(task_id) in TERMINAL

    def in_status(self, *wanted: str) -> List[str]:
        return sorted(tid for tid, s in self.status.items() if s in wanted)

    def unfinished(self) -> List[str]:
        return sorted(tid for tid, s in self.status.items() if s not in TERMINAL)

    def is_complete(self) -> bool:
        return all(s in TERMINAL for s in self.status.values())

    def counts(self) -> Dict[str, int]:
        out: Dict[str, int] = {}
        for s in self.status.values():
            out[s] = out.get(s, 0) + 1
        return out

flow/task.py: """Task definitions."""
from dataclasses import dataclass, field
from typing import FrozenSet, Optional, Tuple

from .retry import RetryPolicy


@dataclass(frozen=True)
class ResourceRequest:
    """How much of a named pool a task holds while it runs."""
    name: str
    amount: int = 1

    def __post_init__(self):
        if self.amount <= 0:
            raise ValueError("resource amount must be positive")


@dataclass(frozen=True)
class Task:
    """One unit of work in the graph.

    `priority` is highest-first. Ties are broken by task id so that a schedule
    is reproducible across runs -- two tasks of equal priority must not swap
    places depending on dictionary ordering.
    """
    task_id: str
    depends_on: FrozenSet[str] = frozenset()
    priority: int = 0
    resources: Tuple[ResourceRequest, ...] = ()
    retry: RetryPolicy = field(default_factory=lambda: RetryPolicy())
    condition: Optional[str] = None

    def __post_init__(self):
        if not self.task_id:
            raise ValueError("task_id must not be empty")
        if self.task_id in self.depends_on:
            raise ValueError(f"{self.task_id} depends on itself")

    def sort_key(self) -> Tuple[int, str]:
        return (-self.priority, self.task_id)

    def requires(self, name: str) -> int:
        """Units of `name` this task needs, zero when it needs none."""
        for request in self.resources:
            if request.name == name:
                return request.amount
        return 0

flow/validation.py: """Static checks run before a graph is executed."""
from typing import List, Mapping

from .expressions import names_in
from .graph import TaskGraph


def check_dependencies(graph: TaskGraph) -> List[str]:
    """Every dependency must name a task that exists."""
    problems: List[str] = []
    known = set(graph.ids())
    for task in graph:
        for upstream in sorted(task.depends_on):
            if upstream not in known:
                problems.append(f"{task.task_id} depends on unknown {upstream}")
    return problems


def check_resources(graph: TaskGraph, capacity: Mapping[str, int]) -> List[str]:
    """A task must not ask for more of a pool than the pool can ever hold.

    A request larger than total capacity can never be admitted, so the task
    would wait forever rather than fail -- worth catching before the run.
    """
    problems: List[str] = []
    for task in graph:
        for request in task.resources:
            if request.name not in capacity:
                problems.append(f"{task.task_id} wants unknown resource "
                                f"{request.name}")
            elif request.amount > capacity[request.name]:
                problems.append(f"{task.task_id} wants {request.amount} of "
                                f"{request.name} but capacity is "
                                f"{capacity[request.name]}")
    return problems


def check_conditions(graph: TaskGraph) -> List[str]:
    """A condition may only reference tasks upstream of it.

    Referring to anything else would read a result that may not have been
    computed when the condition is evaluated.
    """
    problems: List[str] = []
    for task in graph:
        if not task.condition:
            continue
        upstream = graph.ancestors_of(task.task_id)
        for name in names_in(task.condition):
            if name not in upstream:
                problems.append(f"{task.task_id} condition references {name}, "
                                f"which is not upstream of it")
    return problems


def validate_all(graph: TaskGraph, capacity: Mapping[str, int]) -> List[str]:
    """Run every check. Raises for structural faults, returns soft problems."""
    graph.validate()
    return (check_dependencies(graph)
            + check_resources(graph, capacity)
            + check_conditions(graph))


Hidden tests:
tests/test_batching.py: 
import unittest

from flow import ResourcePool, ResourceRequest, Scheduler, Task, TaskGraph


def sched(capacity, tasks, **kw):
    return Scheduler(TaskGraph(tasks), ResourcePool(capacity), **kw)


class TestBatchIsStartable(unittest.TestCase):
    def test_batch_fits_within_capacity(self):
        s = sched({"cpu": 3},
                  [Task("a", resources=(ResourceRequest("cpu", 2),)),
                   Task("b", resources=(ResourceRequest("cpu", 2),))])
        self.assertEqual(s.next_batch(s.new_state()), ["a"])

    def test_whole_batch_can_actually_start(self):
        s = sched({"cpu": 5},
                  [Task("a", resources=(ResourceRequest("cpu", 2),)),
                   Task("b", resources=(ResourceRequest("cpu", 2),)),
                   Task("c", resources=(ResourceRequest("cpu", 2),))])
        state = s.new_state()
        batch = s.next_batch(state)
        for task_id in batch:
            s.start(task_id, state)          # must not raise
        self.assertEqual(len(batch), 2)

    def test_three_way_split(self):
        s = sched({"slots": 4},
                  [Task(name, resources=(ResourceRequest("slots", 3),))
                   for name in ("a", "b", "c")])
        self.assertEqual(s.next_batch(s.new_state()), ["a"])


class TestBatchingOtherwiseUnchanged(unittest.TestCase):
    def test_resourceless_tasks_all_admitted(self):
        s = sched({}, [Task("a"), Task("b"), Task("c")])
        self.assertEqual(s.next_batch(s.new_state()), ["a", "b", "c"])

    def test_max_parallel_still_caps(self):
        s = sched({}, [Task("a"), Task("b"), Task("c")], max_parallel=2)
        self.assertEqual(len(s.next_batch(s.new_state())), 2)

    def test_priority_order_preserved(self):
        s = sched({}, [Task("low", priority=0), Task("high", priority=9)])
        self.assertEqual(s.next_batch(s.new_state()), ["high", "low"])

    def test_everything_fits_when_capacity_is_ample(self):
        s = sched({"cpu": 10},
                  [Task("a", resources=(ResourceRequest("cpu", 1),)),
                   Task("b", resources=(ResourceRequest("cpu", 1),))])
        self.assertEqual(s.next_batch(s.new_state()), ["a", "b"])


Reference fix:
README.md: # flow

A small workflow engine: directed graphs of tasks, run state, and a scheduler
that decides what may run next.

```python
from flow import Task, TaskGraph, ResourcePool, Scheduler

graph = TaskGraph([
    Task("fetch"),
    Task("build", depends_on=frozenset({"fetch"})),
    Task("test",  depends_on=frozenset({"build"})),
])
scheduler = Scheduler(graph, ResourcePool({}))
state = scheduler.new_state()

for task_id in scheduler.next_batch(state):
    scheduler.start(task_id, state)
    scheduler.finish(task_id, state, result=True)
```

## Layout

| module | role |
|---|---|
| `task.py` | `Task`, `ResourceRequest` — what a unit of work declares |
| `graph.py` | `TaskGraph` — dependency structure, traversal, topological order |
| `state.py` | `RunState` — per-task status, the legal transitions between them |
| `scheduler.py` | eligibility, batching, and the start/finish/fail transitions |
| `resources.py` | `ResourcePool` — named capacities held for the duration of a task |
| `retry.py` | `RetryPolicy` — attempt counting and deterministic backoff |
| `calendar.py` | `Calendar`, `Window` — when tasks are permitted to start |
| `expressions.py` | the condition mini-language used to gate tasks |
| `validation.py` | static checks run before a graph executes |
| `serialize.py` | lossless persistence of a run in progress |
| `metrics.py` | counters and an event timeline |
| `clock.py` | injectable time source, so runs are reproducible |

## Design notes

**Determinism.** Anything that could vary between runs is pinned. Ties in
priority are broken by task id, topological order sorts its ready set, and
retry backoff carries no jitter. Two runs of the same graph make the same
decisions in the same order.

**The scheduler does not execute.** It reports what is eligible; the caller
runs the work and reports back. Admission rules are therefore testable without
running anything.

**Half-open intervals.** A calendar window is `[start, end)` throughout, so
adjacent windows neither overlap nor leave a gap.

**Failure propagates.** A task that will never succeed skips everything
downstream of it, because those tasks depend on a result that will not exist.

flow/__init__.py: """A small workflow engine: graphs of tasks, run state, and a scheduler."""
from .calendar import Calendar, Window
from .clock import Clock
from .errors import (CyclicGraph, DuplicateTask, ExpressionError, FlowError,
                     GraphError, IllegalTransition, ResourceExhausted,
                     SerializationError, UnknownResource, UnknownTask)
from .expressions import evaluate, names_in, tokenize
from .graph import TaskGraph
from .metrics import Metrics
from .resources import ResourcePool
from .retry import NO_RETRY, RetryPolicy
from .scheduler import Scheduler
from .serialize import dump_state, load_state, round_trip
from .state import RunState
from .task import ResourceRequest, Task
from .validation import validate_all

__all__ = [
    "Calendar", "Window", "Clock", "TaskGraph", "Metrics", "ResourcePool",
    "RetryPolicy", "NO_RETRY", "Scheduler", "RunState", "Task",
    "ResourceRequest", "validate_all", "evaluate", "names_in", "tokenize",
    "dump_state", "load_state", "round_trip",
    "FlowError", "GraphError", "CyclicGraph", "DuplicateTask", "UnknownTask",
    "IllegalTransition", "ResourceExhausted", "UnknownResource",
    "ExpressionError", "SerializationError",
]

flow/calendar.py: """Scheduling windows.

A window is a half-open interval [start, end): a task may begin exactly at
`start`, and may not begin at `end`. Half-open intervals are used throughout so
that back-to-back windows neither overlap nor leave a gap.
"""
from dataclasses import dataclass
from typing import List, Optional, Sequence


@dataclass(frozen=True)
class Window:
    start: float
    end: float

    def __post_init__(self):
        if self.end <= self.start:
            raise ValueError("window end must be after its start")

    def contains(self, when: float) -> bool:
        return self.start <= when < self.end

    def __str__(self) -> str:
        return f"[{self.start}, {self.end})"


class Calendar:
    """The times at which tasks are permitted to start.

    An empty calendar permits everything; that is the common case and avoids
    forcing every caller to describe an always-open schedule.
    """

    def __init__(self, windows: Optional[Sequence[Window]] = None):
        self.windows: List[Window] = sorted(windows or (),
                                            key=lambda w: (w.start, w.end))

    def is_open(self, when: float) -> bool:
        if not self.windows:
            return True
        return any(window.contains(when) for window in self.windows)

    def next_open(self, when: float) -> Optional[float]:
        """The earliest permitted time at or after `when`."""
        if not self.windows:
            return when
        if self.is_open(when):
            return when
        for window in self.windows:
            if window.start >= when:
                return window.start
        return None

    def closes_at(self, when: float) -> Optional[float]:
        for window in self.windows:
            if window.contains(when):
                return window.end
        return None

flow/clock.py: """Time source.

Everything that needs the current time takes a Clock, so a run can be replayed
deterministically in tests instead of depending on the wall clock.
"""


class Clock:
    """Monotonic seconds since the run started."""

    def __init__(self, start: float = 0.0):
        self._now = float(start)

    def now(self) -> float:
        return self._now

    def advance(self, seconds: float) -> float:
        if seconds < 0:
            raise ValueError("cannot move a clock backwards")
        self._now += float(seconds)
        return self._now

flow/errors.py: """Exception hierarchy for the workflow engine."""


class FlowError(Exception):
    """Base class for everything this package raises."""


class GraphError(FlowError):
    """The task graph is malformed."""


class CyclicGraph(GraphError):
    def __init__(self, cycle):
        super().__init__("cycle: " + " -> ".join(cycle))
        self.cycle = list(cycle)


class UnknownTask(FlowError):
    def __init__(self, task_id: str):
        super().__init__(f"unknown task: {task_id}")
        self.task_id = task_id


class DuplicateTask(GraphError):
    def __init__(self, task_id: str):
        super().__init__(f"task already defined: {task_id}")
        self.task_id = task_id


class IllegalTransition(FlowError):
    def __init__(self, task_id: str, old: str, new: str):
        super().__init__(f"{task_id}: cannot move from {old} to {new}")
        self.task_id = task_id
        self.old = old
        self.new = new


class ResourceExhausted(FlowError):
    def __init__(self, name: str, want: int, free: int):
        super().__init__(f"resource {name!r}: wanted {want}, {free} free")
        self.name = name
        self.want = want
        self.free = free


class UnknownResource(FlowError):
    def __init__(self, name: str):
        super().__init__(f"unknown resource: {name}")
        self.name = name


class ExpressionError(FlowError):
    """A condition expression could not be parsed or evaluated."""


class SerializationError(FlowError):
    """A persisted run could not be restored."""

flow/expressions.py: """A very small condition language for gating tasks.

Grammar, loosest binding first:

    expr     := or_expr
    or_expr  := and_expr ( 'or' and_expr )*
    and_expr := unary ( 'and' unary )*
    unary    := 'not' unary | atom
    atom     := '(' expr ')' | NAME

A NAME resolves against a mapping of facts, so `build and not skip_tests` is
true when `build` is truthy and `skip_tests` is not. `or` binds more loosely
than `and`, as it does in Python.
"""
import re
from typing import List, Mapping, Optional

from .errors import ExpressionError

_TOKEN = re.compile(r"\s*(\(|\)|[A-Za-z_][A-Za-z0-9_.]*)")
_KEYWORDS = ("and", "or", "not")


def tokenize(text: str) -> List[str]:
    tokens: List[str] = []
    pos = 0
    while pos < len(text):
        if text[pos].isspace():
            pos += 1
            continue
        match = _TOKEN.match(text, pos)
        if not match:
            raise ExpressionError(f"unexpected character {text[pos]!r} at {pos}")
        tokens.append(match.group(1))
        pos = match.end()
    return tokens


def names_in(text: str) -> List[str]:
    """Every fact name an expression refers to."""
    return sorted({t for t in tokenize(text)
                   if t not in _KEYWORDS and t not in ("(", ")")})


class _Parser:
    def __init__(self, tokens: List[str], facts: Mapping[str, object]):
        self.tokens = tokens
        self.pos = 0
        self.facts = facts

    def peek(self) -> Optional[str]:
        return self.tokens[self.pos] if self.pos < len(self.tokens) else None

    def take(self) -> str:
        token = self.peek()
        if token is None:
            raise ExpressionError("unexpected end of expression")
        self.pos += 1
        return token

    def parse(self) -> bool:
        value = self.or_expr()
        if self.peek() is not None:
            raise ExpressionError(f"trailing input at {self.peek()!r}")
        return value

    def or_expr(self) -> bool:
        value = self.and_expr()
        while self.peek() == "or":
            self.take()
            right = self.and_expr()
            value = value or right
        return value

    def and_expr(self) -> bool:
        value = self.unary()
        while self.peek() == "and":
            self.take()
            right = self.unary()
            value = value and right
        return value

    def unary(self) -> bool:
        if self.peek() == "not":
            self.take()
            return not self.unary()
        return self.atom()

    def atom(self) -> bool:
        token = self.take()
        if token == "(":
            value = self.or_expr()
            if self.take() != ")":
                raise ExpressionError("missing closing parenthesis")
            return value
        if token in _KEYWORDS or token == ")":
            raise ExpressionError(f"unexpected token {token!r}")
        return bool(self.facts.get(token, False))


def evaluate(expression: Optional[str], facts: Mapping[str, object]) -> bool:
    """Evaluate a condition. An absent or blank condition is always true."""
    if expression is None or not expression.strip():
        return True
    tokens = tokenize(expression)
    if not tokens:
        return True
    return _Parser(tokens, facts).parse()

flow/graph.py: """The task graph: dependency structure and traversal."""
from typing import Dict, Iterable, Iterator, List, Optional, Set

from .errors import CyclicGraph, DuplicateTask, UnknownTask
from .task import Task


class TaskGraph:
    """A directed acyclic graph of tasks, keyed by task id.

    Edges point from a dependency to the task that needs it: if B depends on A
    then A is *upstream* of B, and B is a *dependent* of A.
    """

    def __init__(self, tasks: Optional[Iterable[Task]] = None):
        self._tasks: Dict[str, Task] = {}
        self._dependents: Dict[str, Set[str]] = {}
        for task in tasks or ():
            self.add(task)

    # -- construction ----------------------------------------------------

    def add(self, task: Task) -> Task:
        if task.task_id in self._tasks:
            raise DuplicateTask(task.task_id)
        self._tasks[task.task_id] = task
        self._dependents.setdefault(task.task_id, set())
        for upstream in task.depends_on:
            self._dependents.setdefault(upstream, set()).add(task.task_id)
        return task

    def validate(self) -> None:
        """Check every dependency resolves and the graph is acyclic."""
        for task in self._tasks.values():
            for upstream in task.depends_on:
                if upstream not in self._tasks:
                    raise UnknownTask(upstream)
        self.topological_order()

    # -- access ----------------------------------------------------------

    def __len__(self) -> int:
        return len(self._tasks)

    def __contains__(self, task_id: object) -> bool:
        return task_id in self._tasks

    def __iter__(self) -> Iterator[Task]:
        return iter(self._tasks.values())

    def get(self, task_id: str) -> Task:
        if task_id not in self._tasks:
            raise UnknownTask(task_id)
        return self._tasks[task_id]

    def ids(self) -> List[str]:
        return sorted(self._tasks)

    def upstream_of(self, task_id: str) -> Set[str]:
        """Direct dependencies of `task_id`."""
        return set(self.get(task_id).depends_on)

    def dependents_of(self, task_id: str) -> Set[str]:
        """Tasks that directly depend on `task_id`."""
        if task_id not in self._tasks:
            raise UnknownTask(task_id)
        return set(self._dependents.get(task_id, ()))

    def roots(self) -> List[str]:
        """Tasks with no dependencies, in sorted order."""
        return sorted(t.task_id for t in self._tasks.values() if not t.depends_on)

    def leaves(self) -> List[str]:
        return sorted(tid for tid in self._tasks
                      if not self._dependents.get(tid))

    # -- traversal -------------------------------------------------------

    def descendants_of(self, task_id: str) -> Set[str]:
        """Everything reachable downstream of `task_id`, excluding itself."""
        seen: Set[str] = set()
        stack = list(self.dependents_of(task_id))
        while stack:
            current = stack.pop()
            if current in seen:
                continue
            seen.add(current)
            stack.extend(self._dependents.get(current, ()))
        return seen

    def ancestors_of(self, task_id: str) -> Set[str]:
        """Everything `task_id` transitively depends on."""
        seen: Set[str] = set()
        stack = list(self.upstream_of(task_id))
        while stack:
            current = stack.pop()
            if current in seen:
                continue
            seen.add(current)
            stack.extend(self.get(current).depends_on)
        return seen

    def topological_order(self) -> List[str]:
        """A dependency-respecting order, ties broken by id for determinism."""
        indegree = {tid: len(self._tasks[tid].depends_on) for tid in self._tasks}
        ready = sorted(tid for tid, n in indegree.items() if n == 0)
        order: List[str] = []
        while ready:
            current = ready.pop(0)
            order.append(current)
            for downstream in sorted(self._dependents.get(current, ())):
                indegree[downstream] -= 1
                if indegree[downstream] == 0:
                    ready.append(downstream)
            ready.sort()
        if len(order) != len(self._tasks):
            raise CyclicGraph(sorted(set(self._tasks) - set(order)))
        return order

flow/metrics.py: """Counters and an event timeline collected over a run."""
from typing import Dict, List, Tuple


class Metrics:
    def __init__(self) -> None:
        self.counters: Dict[str, int] = {}
        self.timeline: List[Tuple[float, str, str]] = []

    def bump(self, name: str, by: int = 1) -> int:
        self.counters[name] = self.counters.get(name, 0) + by
        return self.counters[name]

    def record(self, when: float, task_id: str, event: str) -> None:
        self.timeline.append((when, task_id, event))
        self.bump(event)

    def events_for(self, task_id: str) -> List[str]:
        return [event for _, tid, event in self.timeline if tid == task_id]

    def count(self, name: str) -> int:
        return self.counters.get(name, 0)

    def summary(self) -> Dict[str, int]:
        return dict(sorted(self.counters.items()))

flow/resources.py: """Named resource pools with fixed capacity."""
from typing import Dict, Iterable, List, Mapping, Tuple

from .errors import ResourceExhausted, UnknownResource
from .task import ResourceRequest


class ResourcePool:
    """Tracks how much of each named resource is currently held.

    Holdings are recorded per task so that releasing is exact: a task gives
    back precisely what it took, which is what keeps a failed attempt from
    leaking capacity for the rest of the run.
    """

    def __init__(self, capacity: Mapping[str, int]):
        for name, total in capacity.items():
            if total < 0:
                raise ValueError(f"capacity for {name!r} must not be negative")
        self.capacity: Dict[str, int] = dict(capacity)
        self._held: Dict[str, Dict[str, int]] = {}

    def free(self, name: str) -> int:
        if name not in self.capacity:
            raise UnknownResource(name)
        used = sum(held.get(name, 0) for held in self._held.values())
        return self.capacity[name] - used

    def can_admit(self, requests: Iterable[ResourceRequest]) -> bool:
        for request in requests:
            if request.name not in self.capacity:
                raise UnknownResource(request.name)
            if self.free(request.name) < request.amount:
                return False
        return True

    def acquire(self, task_id: str, requests: Iterable[ResourceRequest]) -> None:
        """Take every request at once, or nothing at all."""
        requests = list(requests)
        for request in requests:
            if request.name not in self.capacity:
                raise UnknownResource(request.name)
            if self.free(request.name) < request.amount:
                raise ResourceExhausted(request.name, request.amount,
                                        self.free(request.name))
        holding = self._held.setdefault(task_id, {})
        for request in requests:
            holding[request.name] = holding.get(request.name, 0) + request.amount

    def release(self, task_id: str) -> Dict[str, int]:
        """Give back everything `task_id` holds. Safe to call twice."""
        return self._held.pop(task_id, {})

    def held_by(self, task_id: str) -> Dict[str, int]:
        return dict(self._held.get(task_id, {}))

    def in_use(self) -> Dict[str, int]:
        out: Dict[str, int] = {name: 0 for name in self.capacity}
        for holding in self._held.values():
            for name, amount in holding.items():
                out[name] = out.get(name, 0) + amount
        return out

flow/retry.py: """Retry policies.

Backoff is deterministic on purpose: a scheduler test that depended on jitter
would be flaky, and the engine's own behaviour must be reproducible.
"""
from dataclasses import dataclass


@dataclass(frozen=True)
class RetryPolicy:
    """How many times to retry, and how long to wait between attempts.

    `max_attempts` counts the *total* number of attempts, so the default of 1
    means a task runs once and is not retried.
    """
    max_attempts: int = 1
    base_delay: float = 1.0
    multiplier: float = 2.0
    max_delay: float = 300.0

    def __post_init__(self):
        if self.max_attempts < 1:
            raise ValueError("max_attempts must be at least 1")
        if self.base_delay < 0:
            raise ValueError("base_delay must not be negative")
        if self.multiplier < 1.0:
            raise ValueError("multiplier must be at least 1.0")

    def should_retry(self, attempts_made: int) -> bool:
        """True when another attempt is still allowed."""
        return attempts_made < self.max_attempts

    def delay_for(self, attempts_made: int) -> float:
        """Seconds to wait before attempt number `attempts_made` + 1.

        The first retry waits `base_delay`; each subsequent one multiplies by
        `multiplier`, clamped at `max_delay`.
        """
        if attempts_made < 1:
            return 0.0
        delay = self.base_delay * (self.multiplier ** (attempts_made - 1))
        return min(delay, self.max_delay)


NO_RETRY = RetryPolicy(max_attempts=1)

flow/scheduler.py: """The scheduler: decides what may run, and drives a run to completion."""
from typing import Dict, List, Mapping, Optional, Sequence

from . import state as st
from .calendar import Calendar
from .clock import Clock
from .expressions import evaluate
from .graph import TaskGraph
from .metrics import Metrics
from .resources import ResourcePool
from .task import Task


class Scheduler:
    """Turns a graph plus a run state into a sequence of dispatch decisions.

    The scheduler never executes anything itself. `next_batch` reports what is
    eligible right now, and the caller reports back what happened. That split
    keeps the admission rules testable without running any work.
    """

    def __init__(self, graph: TaskGraph, pool: ResourcePool,
                 calendar: Optional[Calendar] = None,
                 clock: Optional[Clock] = None,
                 metrics: Optional[Metrics] = None,
                 max_parallel: int = 0):
        self.graph = graph
        self.pool = pool
        self.calendar = calendar or Calendar()
        self.clock = clock or Clock()
        self.metrics = metrics or Metrics()
        self.max_parallel = max_parallel

    # -- eligibility -----------------------------------------------------

    def upstream_satisfied(self, task: Task, state: st.RunState) -> bool:
        """True when every dependency has finished successfully.

        A dependency that was skipped does not satisfy anything downstream:
        its result was never produced, so anything needing it is skipped too.
        """
        for upstream in task.depends_on:
            if state.get(upstream) != st.SUCCEEDED:
                return False
        return True

    def upstream_settled(self, task: Task, state: st.RunState) -> bool:
        """True once no dependency can change state again."""
        return all(state.is_terminal(u) for u in task.depends_on)

    def condition_holds(self, task: Task, state: st.RunState) -> bool:
        """Evaluate the task's condition against upstream results."""
        if not task.condition:
            return True
        facts = {tid: state.results.get(tid, False)
                 for tid in self.graph.ancestors_of(task.task_id)
                 if state.get(tid) == st.SUCCEEDED}
        return evaluate(task.condition, facts)

    def running_count(self, state: st.RunState) -> int:
        return len(state.in_status(st.RUNNING))

    def eligible(self, state: st.RunState) -> List[str]:
        """Every task that could be dispatched now, in dispatch order."""
        when = self.clock.now()
        if not self.calendar.is_open(when):
            return []

        candidates: List[Task] = []
        for task_id in self.graph.ids():
            # RETRYING counts as a candidate: its backoff is enforced below by
            # ready_at, and leaving it out would mean a retry never fires.
            if state.get(task_id) not in (st.PENDING, st.READY, st.RETRYING):
                continue
            task = self.graph.get(task_id)
            if not self.upstream_satisfied(task, state):
                continue
            if not self.condition_holds(task, state):
                continue
            earliest = state.ready_at.get(task_id)
            if earliest is not None and when < earliest:
                continue
            candidates.append(task)

        candidates.sort(key=lambda t: t.sort_key())
        return [t.task_id for t in candidates]

    def next_batch(self, state: st.RunState) -> List[str]:
        """The eligible tasks that also fit within resources and parallelism.

        Resource checks are cumulative across the batch: two tasks that each
        fit on their own must not both be admitted when together they exceed
        the pool.
        """
        chosen: List[str] = []
        reserved: Dict[str, int] = {}
        in_flight = self.running_count(state)

        for task_id in self.eligible(state):
            if self.max_parallel and in_flight + len(chosen) >= self.max_parallel:
                break
            task = self.graph.get(task_id)
            fits = True
            for request in task.resources:
                already = reserved.get(request.name, 0)
                if self.pool.free(request.name) - already < request.amount:
                    fits = False
                    break
            if not fits:
                continue
            for request in task.resources:
                reserved[request.name] = reserved.get(request.name, 0) + request.amount
            chosen.append(task_id)
        return chosen

    # -- transitions -----------------------------------------------------

    def start(self, task_id: str, state: st.RunState) -> None:
        task = self.graph.get(task_id)
        if state.get(task_id) == st.PENDING:
            state.set(task_id, st.READY)
        elif state.get(task_id) == st.RETRYING:
            state.set(task_id, st.READY)
        self.pool.acquire(task_id, task.resources)
        state.set(task_id, st.RUNNING)
        state.record_attempt(task_id)
        self.metrics.record(self.clock.now(), task_id, "started")

    def finish(self, task_id: str, state: st.RunState, result: object = None) -> None:
        self.pool.release(task_id)
        state.succeed(task_id, result)
        self.metrics.record(self.clock.now(), task_id, "succeeded")

    def fail(self, task_id: str, state: st.RunState) -> str:
        """Record a failed attempt, retrying it if the policy still allows.

        The pool is released on every failure, not only the last one: a task
        that is going to be retried must not keep holding capacity while it
        waits out its backoff.
        """
        self.pool.release(task_id)
        task = self.graph.get(task_id)
        attempts = state.attempts.get(task_id, 0)
        if task.retry.should_retry(attempts):
            state.set(task_id, st.RETRYING)
            delay = task.retry.delay_for(attempts)
            state.ready_at[task_id] = self.clock.now() + delay
            self.metrics.record(self.clock.now(), task_id, "retrying")
            return st.RETRYING
        state.set(task_id, st.FAILED)
        self.metrics.record(self.clock.now(), task_id, "failed")
        self.cascade_skip(task_id, state)
        return st.FAILED

    def cascade_skip(self, task_id: str, state: st.RunState) -> List[str]:
        """Skip everything downstream of a task that will never succeed."""
        skipped: List[str] = []
        for downstream in sorted(self.graph.descendants_of(task_id)):
            if state.get(downstream) in (st.PENDING, st.READY):
                state.set(downstream, st.SKIPPED)
                self.metrics.record(self.clock.now(), downstream, "skipped")
                skipped.append(downstream)
        return skipped

    def settle_unreachable(self, state: st.RunState) -> List[str]:
        """Skip tasks whose dependencies are settled but not all successful."""
        skipped: List[str] = []
        for task_id in state.in_status(st.PENDING, st.READY):
            task = self.graph.get(task_id)
            if not self.upstream_settled(task, state):
                continue
            if not self.upstream_satisfied(task, state):
                state.set(task_id, st.SKIPPED)
                skipped.append(task_id)
            elif not self.condition_holds(task, state):
                state.set(task_id, st.SKIPPED)
                skipped.append(task_id)
        return skipped

    def new_state(self) -> st.RunState:
        return st.RunState(self.graph.ids())

flow/serialize.py: """Persist and restore a run in progress.

A round trip must be lossless: a run reloaded from disk has to make exactly
the decisions it would have made had it never stopped. Anything the scheduler
reads therefore has to be written.
"""
from typing import Any, Dict

from .errors import SerializationError
from .state import RunState

FORMAT_VERSION = 2


def dump_state(state: RunState) -> Dict[str, Any]:
    return {
        "version": FORMAT_VERSION,
        "status": dict(state.status),
        "attempts": dict(state.attempts),
        "results": dict(state.results),
        "ready_at": dict(state.ready_at),
    }


def load_state(payload: Dict[str, Any]) -> RunState:
    if not isinstance(payload, dict):
        raise SerializationError("payload must be a mapping")
    version = payload.get("version")
    if version != FORMAT_VERSION:
        raise SerializationError(
            f"unsupported format version {version!r}, expected {FORMAT_VERSION}")
    status = payload.get("status")
    if not isinstance(status, dict):
        raise SerializationError("missing status map")

    state = RunState(status)
    state.status = dict(status)
    state.attempts = dict(payload.get("attempts") or {})
    state.results = dict(payload.get("results") or {})
    state.ready_at = dict(payload.get("ready_at") or {})
    for task_id in state.status:
        state.attempts.setdefault(task_id, 0)
    return state


def round_trip(state: RunState) -> RunState:
    return load_state(dump_state(state))

flow/state.py: """Run state: the status of every task in one execution of a graph."""
from typing import Dict, Iterable, List, Optional, Set

from .errors import IllegalTransition, UnknownTask

PENDING = "pending"
READY = "ready"
RUNNING = "running"
SUCCEEDED = "succeeded"
FAILED = "failed"
RETRYING = "retrying"
SKIPPED = "skipped"

TERMINAL = frozenset({SUCCEEDED, FAILED, SKIPPED})

# Which moves the engine permits. A task that has finished stays finished:
# re-running one would silently invalidate everything computed downstream of it.
ALLOWED: Dict[str, Set[str]] = {
    PENDING: {READY, SKIPPED},
    READY: {RUNNING, SKIPPED, PENDING},
    RUNNING: {SUCCEEDED, FAILED, RETRYING},
    RETRYING: {READY, FAILED},
    SUCCEEDED: set(),
    FAILED: set(),
    SKIPPED: set(),
}


class RunState:
    """Mutable status of each task, plus attempt counts and results."""

    def __init__(self, task_ids: Iterable[str]):
        self.status: Dict[str, str] = {tid: PENDING for tid in task_ids}
        self.attempts: Dict[str, int] = {tid: 0 for tid in self.status}
        self.results: Dict[str, object] = {}
        self.ready_at: Dict[str, float] = {}

    def _check(self, task_id: str) -> str:
        if task_id not in self.status:
            raise UnknownTask(task_id)
        return self.status[task_id]

    def get(self, task_id: str) -> str:
        return self._check(task_id)

    def set(self, task_id: str, new: str) -> str:
        old = self._check(task_id)
        if new not in ALLOWED.get(old, set()):
            raise IllegalTransition(task_id, old, new)
        self.status[task_id] = new
        return new

    def record_attempt(self, task_id: str) -> int:
        self._check(task_id)
        self.attempts[task_id] = self.attempts.get(task_id, 0) + 1
        return self.attempts[task_id]

    def succeed(self, task_id: str, result: object = None) -> None:
        self.set(task_id, SUCCEEDED)
        self.results[task_id] = result

    def is_terminal(self, task_id: str) -> bool:
        return self._check(task_id) in TERMINAL

    def in_status(self, *wanted: str) -> List[str]:
        return sorted(tid for tid, s in self.status.items() if s in wanted)

    def unfinished(self) -> List[str]:
        return sorted(tid for tid, s in self.status.items() if s not in TERMINAL)

    def is_complete(self) -> bool:
        return all(s in TERMINAL for s in self.status.values())

    def counts(self) -> Dict[str, int]:
        out: Dict[str, int] = {}
        for s in self.status.values():
            out[s] = out.get(s, 0) + 1
        return out

flow/task.py: """Task definitions."""
from dataclasses import dataclass, field
from typing import FrozenSet, Optional, Tuple

from .retry import RetryPolicy


@dataclass(frozen=True)
class ResourceRequest:
    """How much of a named pool a task holds while it runs."""
    name: str
    amount: int = 1

    def __post_init__(self):
        if self.amount <= 0:
            raise ValueError("resource amount must be positive")


@dataclass(frozen=True)
class Task:
    """One unit of work in the graph.

    `priority` is highest-first. Ties are broken by task id so that a schedule
    is reproducible across runs -- two tasks of equal priority must not swap
    places depending on dictionary ordering.
    """
    task_id: str
    depends_on: FrozenSet[str] = frozenset()
    priority: int = 0
    resources: Tuple[ResourceRequest, ...] = ()
    retry: RetryPolicy = field(default_factory=lambda: RetryPolicy())
    condition: Optional[str] = None

    def __post_init__(self):
        if not self.task_id:
            raise ValueError("task_id must not be empty")
        if self.task_id in self.depends_on:
            raise ValueError(f"{self.task_id} depends on itself")

    def sort_key(self) -> Tuple[int, str]:
        return (-self.priority, self.task_id)

    def requires(self, name: str) -> int:
        """Units of `name` this task needs, zero when it needs none."""
        for request in self.resources:
            if request.name == name:
                return request.amount
        return 0

flow/validation.py: """Static checks run before a graph is executed."""
from typing import List, Mapping

from .expressions import names_in
from .graph import TaskGraph


def check_dependencies(graph: TaskGraph) -> List[str]:
    """Every dependency must name a task that exists."""
    problems: List[str] = []
    known = set(graph.ids())
    for task in graph:
        for upstream in sorted(task.depends_on):
            if upstream not in known:
                problems.append(f"{task.task_id} depends on unknown {upstream}")
    return problems


def check_resources(graph: TaskGraph, capacity: Mapping[str, int]) -> List[str]:
    """A task must not ask for more of a pool than the pool can ever hold.

    A request larger than total capacity can never be admitted, so the task
    would wait forever rather than fail -- worth catching before the run.
    """
    problems: List[str] = []
    for task in graph:
        for request in task.resources:
            if request.name not in capacity:
                problems.append(f"{task.task_id} wants unknown resource "
                                f"{request.name}")
            elif request.amount > capacity[request.name]:
                problems.append(f"{task.task_id} wants {request.amount} of "
                                f"{request.name} but capacity is "
                                f"{capacity[request.name]}")
    return problems


def check_conditions(graph: TaskGraph) -> List[str]:
    """A condition may only reference tasks upstream of it.

    Referring to anything else would read a result that may not have been
    computed when the condition is evaluated.
    """
    problems: List[str] = []
    for task in graph:
        if not task.condition:
            continue
        upstream = graph.ancestors_of(task.task_id)
        for name in names_in(task.condition):
            if name not in upstream:
                problems.append(f"{task.task_id} condition references {name}, "
                                f"which is not upstream of it")
    return problems


def validate_all(graph: TaskGraph, capacity: Mapping[str, int]) -> List[str]:
    """Run every check. Raises for structural faults, returns soft problems."""
    graph.validate()
    return (check_dependencies(graph)
            + check_resources(graph, capacity)
            + check_conditions(graph))