orchestrata_ultimate.core.threading.test_scheduler

Reference › orchestrata_ultimate › core › threading › test_scheduler

A constraint-based scheduler for test modules (section 9, row B28). It turns a scheduling problem into a Prolog term, solves it with the CLP(FD) module test_schedule in prolog/test_schedule.pl through PrologEngine, and reads the answer back. The inputs are per-module durations, worker capability sets, per-module infrastructure requirements, conflict pairs that may not share a worker, and an optional wall-clock budget. No requirement ids exist for this row, so the module is deliberately untraced. It is used by scripts/schedule_tests.py and is not wired into the test runner; the module says that wiring is a separate step.

plan_schedule returns a frozen ScheduledPlan: module-to-worker assignments (workers are identified by list position), the predicted makespan in milliseconds, and a feasible flag, which is false only when a budget is given and the best plan still exceeds it. It always solves for the unconstrained optimum, which is equivalent to and cheaper than a budgeted solve, because the optimum fits the budget exactly when any plan does. Up to an exact-solve limit the problem is solved exactly (_exact_schedule); above it _decomposed_schedule solves the heaviest modules exactly and places the rest greedily heaviest-first (_place_greedy), because the exact model exhausts the Prolog global stack at corpus scale, so the result is then optimal only within the exact phase. Malformed input (no workers, a module with no eligible worker, or one barred from every eligible worker by conflicts) raises ValueError naming the module. load_durations and load_conflicts read JSON inputs, and the Prolog file is consulted once per process under a lock.

At a glance

Class diagram

Dependency diagram

API

Classes

ScheduledPlan (dataclass)

Decorators: dataclass(frozen=True) · Line 50

The result of a scheduling attempt.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/core/threading/test_scheduler.py

"""CLP(FD) test-module scheduling (§9 row B28).

Marshals a scheduling problem (per-module durations, per-worker infra
capacity, module-conflict pairs, an optional wall-clock budget) to the
``test_schedule`` Prolog module (``prolog/test_schedule.pl``) via the
canonical janus adapter (``core.prolog_adapter.PrologEngine``) and back.

No REQ ids exist for this §9 row (B28) -- ``@traces`` is intentionally
omitted throughout this module.

DESIGN NOTE (schedule/5 vs best_schedule/4): ``plan_schedule`` below always
solves the UNCONSTRAINED optimum via ``best_schedule_pairs/4`` rather than
calling ``schedule_pairs/5`` with ``budget_ms``. This is mathematically
equivalent and cheaper (one CLP(FD) solve instead of two): the global
optimum makespan is achievable if and only if it is <= budget_ms -- if it
is, that same optimal assignment already satisfies the budget; if it is
not, no assignment could (``schedule/5`` searches a subset of the same
space with an extra ``Makespan #=< Budget`` constraint, so it can only
ever match or lose to the unconstrained optimum). ``schedule/5`` itself is
still fully implemented, exported, and exercised directly by
``tests/prolog/test_test_schedule_pl.py`` per this task's spec -- this
module just does not need to call it.

ADOPTION SEAM: this module is NOT wired into ``run_tests.py`` /
``run_tests_threadpool.py`` in this task (those are load-bearing and
out of scope here) -- see ``scripts/schedule_tests.py``'s "B28 adoption
seam" note for what wiring would look like.
"""

from __future__ import annotations

import json
import threading
from dataclasses import dataclass
from typing import Any

from orchestrata_ultimate.core.paths import get_prolog_dir
from orchestrata_ultimate.core.prolog_adapter import PrologEngine

# Guards a process-global, consult-once load of test_schedule.pl -- the
# janus SWI engine/module table is process-global regardless of which
# PrologEngine instance calls consult() (see core.prolog_adapter's
# _EXPLAIN_LOADED for the identical, already-established pattern this
# mirrors).
_SCHEDULE_LOADED = False
_SCHEDULE_LOCK = threading.Lock()


@dataclass(frozen=True)
class ScheduledPlan:
    """The result of a scheduling attempt.

    Attributes:
        assignments: module path -> worker index (0-based, positional into
            the ``workers`` list ``plan_schedule`` was called with).
        predicted_makespan_ms: the slowest worker's total assigned duration,
            in milliseconds, under ``assignments``.
        feasible: True when no ``budget_ms`` was given, or the achieved
            makespan is within it. False means ``assignments`` is the best
            achievable (unconstrained-optimum) fallback plan, which still
            exceeds the requested budget.
    """

    assignments: dict[str, int]
    predicted_makespan_ms: int
    feasible: bool


def _ensure_schedule_pl_loaded() -> None:
    """Consult test_schedule.pl into the process-global janus engine, once."""
    global _SCHEDULE_LOADED
    if _SCHEDULE_LOADED:
        return
    with _SCHEDULE_LOCK:
        if _SCHEDULE_LOADED:
            return
        engine = PrologEngine()
        engine.consult(str(get_prolog_dir() / "test_schedule.pl"))
        _SCHEDULE_LOADED = True


def _atom(value: str) -> str:
    """Render a Python string as a GROUND, quoted Prolog atom.

    Mirrors ``postman.governance._to_prolog_term``'s string-escaping rule
    (single-quoted, backslash/quote-escaped) -- there is no shared helper
    because that module lives in the postman package, out of this task's
    file scope.
    """
    escaped = value.replace("\\", "\\\\").replace("'", "\\'")
    return f"'{escaped}'"


def _infra_term(infra: frozenset[str]) -> str:
    """Render an infra-capability/requirement set as a Prolog atom list."""
    return "[" + ",".join(_atom(x) for x in sorted(infra)) + "]"


def _tasks_term(
    durations: dict[str, int], module_infra: dict[str, frozenset[str]]
) -> str:
    parts = [
        f"task({_atom(module)}, {int(duration)}, "
        f"{_infra_term(module_infra.get(module, frozenset()))})"
        for module, duration in durations.items()
    ]
    return "[" + ",".join(parts) + "]"


def _workers_term(workers: list[frozenset[str]]) -> str:
    """Render workers as ``worker(Idx, InfraCaps)`` -- Idx is the worker's
    0-based position in ``workers``, which doubles as its Python-facing
    ``ScheduledPlan.assignments`` value."""
    parts = [f"worker({idx}, {_infra_term(caps)})" for idx, caps in enumerate(workers)]
    return "[" + ",".join(parts) + "]"


def _conflicts_term(conflicts: set[frozenset[str]], known_modules: set[str]) -> str:
    """Render conflicts as ``Id1-Id2`` pairs, restricted to pairs where BOTH
    modules are part of this scheduling run -- a conflicts file may name a
    broader/stale set of modules than the current ``durations``."""
    parts = []
    for pair in conflicts:
        a, b = sorted(pair)
        if a not in known_modules or b not in known_modules:
            continue
        parts.append(f"{_atom(a)}-{_atom(b)}")
    return "[" + ",".join(parts) + "]"


def _best_schedule(
    tasks_term: str, workers_term: str, conflicts_term: str
) -> dict[str, int]:
    """Call ``test_schedule:best_schedule_pairs/4`` and return module ->
    worker-index. Raises RuntimeError if it finds no solution -- with a
    nonempty, eligibility-validated ``tasks_term`` this should always
    succeed (see ``build_model``'s NumWorkers > 0 / eligibility guards in
    test_schedule.pl); a failure here means the eligibility pre-check in
    ``plan_schedule`` missed a case, i.e. a bug, not a normal outcome.
    """
    _ensure_schedule_pl_loaded()
    engine = PrologEngine()
    goal = (
        f"test_schedule:best_schedule_pairs({tasks_term}, {workers_term}, "
        f"{conflicts_term}, Pairs)"
    )
    result = engine.query_once(goal)
    pairs = result.get("Pairs") if result is not None else None
    if pairs is None:
        raise RuntimeError(
            "best_schedule found no solution for a nonempty, "
            "eligibility-validated task set -- this indicates a bug"
        )
    return {str(task_id): int(worker_idx) for task_id, worker_idx in pairs}


def _validate_eligibility(
    durations: dict[str, int],
    module_infra: dict[str, frozenset[str]],
    workers: list[frozenset[str]],
) -> None:
    """Raise ValueError, naming the module, for the first module in
    ``durations`` with no worker whose infra capabilities are a superset of
    its requirements."""
    for module in durations:
        reqs = module_infra.get(module, frozenset())
        if not any(reqs <= worker_caps for worker_caps in workers):
            raise ValueError(
                f"plan_schedule: module {module!r} has no eligible worker "
                f"(requires {sorted(reqs)}; none of the {len(workers)} "
                "worker(s) satisfy it)"
            )


def _makespan_of(assignments: dict[str, int], durations: dict[str, int]) -> int:
    """The slowest worker's total assigned duration under ``assignments``."""
    per_worker_load: dict[int, int] = {}
    for module, worker_idx in assignments.items():
        per_worker_load[worker_idx] = (
            per_worker_load.get(worker_idx, 0) + durations[module]
        )
    return max(per_worker_load.values(), default=0)


# Exact CLP(FD) solving is superlinear and then falls off a cliff: 25 modules
# solve in 7.35s, 50 exceed 90s, and the full 1,674-module corpus exhausts the
# 1 GB global stack during PROPAGATION (clpfd:sum_finite_domains over a
# 1,708-element list, depth 1,012, 399 choice points -- before search begins).
# Above this many modules the problem is decomposed instead. Tunable per call
# via plan_schedule(..., exact_solve_limit=N) so the tests can pin both paths.
_EXACT_SOLVE_LIMIT = 24


def _exact_schedule(
    durations: dict[str, int],
    workers: list[frozenset[str]],
    module_infra: dict[str, frozenset[str]],
    conflicts: set[frozenset[str]],
) -> dict[str, int]:
    """Solve this (small) task set exactly, via the CLP(FD) model."""
    tasks_term = _tasks_term(durations, module_infra)
    workers_term = _workers_term(workers)
    conflicts_term = _conflicts_term(conflicts, set(durations))
    return _best_schedule(tasks_term, workers_term, conflicts_term)


def _by_weight(durations: dict[str, int]) -> list[str]:
    """Modules heaviest-first; ties broken by name so a plan is reproducible."""
    return sorted(durations, key=lambda m: (-durations[m], m))


def _conflict_partners(conflicts: set[frozenset[str]]) -> dict[str, set[str]]:
    """module -> the modules it may not share a worker with."""
    partners: dict[str, set[str]] = {}
    for pair in conflicts:
        members = tuple(pair)
        if len(members) != 2:
            continue  # a self-pair constrains nothing
        a, b = members
        partners.setdefault(a, set()).add(b)
        partners.setdefault(b, set()).add(a)
    return partners


def _place_greedy(
    modules: list[str],
    assignments: dict[str, int],
    durations: dict[str, int],
    workers: list[frozenset[str]],
    module_infra: dict[str, frozenset[str]],
    conflicts: set[frozenset[str]],
) -> None:
    """Place `modules` onto `workers` by LPT, mutating `assignments`.

    `assignments` may already hold the exact phase's placements; their loads
    and conflict footprints are carried in, so a pair straddling the two
    phases is still honoured. Each module goes to the least-loaded worker that
    both satisfies its infra requirements and holds nothing it conflicts with.
    """
    partners = _conflict_partners(conflicts)

    loads = [0 for _ in workers]
    blocked: list[set[str]] = [set() for _ in workers]
    for module, idx in assignments.items():
        loads[idx] += durations[module]
        blocked[idx] |= partners.get(module, set())

    for module in modules:
        reqs = module_infra.get(module, frozenset())
        best: int | None = None
        for idx, caps in enumerate(workers):
            if not reqs <= caps or module in blocked[idx]:
                continue
            if best is None or loads[idx] < loads[best]:
                best = idx
        if best is None:
            raise ValueError(
                f"plan_schedule: module {module!r} cannot be placed -- every "
                f"worker whose infra satisfies it already holds a module it "
                f"conflicts with"
            )
        assignments[module] = best
        loads[best] += durations[module]
        blocked[best] |= partners.get(module, set())


def _decomposed_schedule(
    durations: dict[str, int],
    workers: list[frozenset[str]],
    module_infra: dict[str, frozenset[str]],
    conflicts: set[frozenset[str]],
    exact_solve_limit: int,
) -> dict[str, int]:
    """Exact-solve the heaviest `exact_solve_limit` modules, LPT the rest.

    The makespan is decided by the heavy tail, so that is where the exact
    solver is spent; the light remainder perturbs it little and LPT is the
    standard heuristic for it.
    """
    ordered = _by_weight(durations)
    heavy, light = ordered[:exact_solve_limit], ordered[exact_solve_limit:]

    heavy_set = set(heavy)
    heavy_durations = {m: durations[m] for m in heavy}
    # Only pairs wholly inside the heavy set can be modelled here; a straddling
    # pair is enforced by _place_greedy when its light member is placed.
    heavy_conflicts = {p for p in conflicts if set(p) <= heavy_set}

    assignments = _exact_schedule(
        heavy_durations, workers, module_infra, heavy_conflicts
    )
    _place_greedy(light, assignments, durations, workers, module_infra, conflicts)
    return assignments


def plan_schedule(
    durations: dict[str, int],
    workers: list[frozenset[str]],
    module_infra: dict[str, frozenset[str]],
    conflicts: set[frozenset[str]],
    budget_ms: int | None,
    *,
    exact_solve_limit: int = _EXACT_SOLVE_LIMIT,
) -> ScheduledPlan:
    """Plan a budget-aware (or plain makespan-minimal) test-module schedule.

    ``workers``' identity is its 0-based position in the list; a module
    absent from ``module_infra`` is treated as requiring nothing;
    ``budget_ms=None`` just minimizes makespan with no budget check.
    ``feasible`` is False only when ``budget_ms`` is given and even the
    achieved assignment exceeds it -- ``assignments``/
    ``predicted_makespan_ms`` are still that plan, never omitted.

    Up to ``exact_solve_limit`` modules are solved exactly by the CLP(FD)
    model. Above it the problem is DECOMPOSED -- the heaviest
    ``exact_solve_limit`` modules are solved exactly and the remainder placed
    by LPT -- because the exact model does not merely get slow at corpus
    scale, it exhausts the Prolog global stack during propagation before
    search begins (see ``_EXACT_SOLVE_LIMIT``). The result is then
    makespan-minimal only within the exact phase, not globally.

    Raises ValueError on malformed input -- no workers given, a module with
    no eligible worker, or a module every eligible worker is barred from by
    conflicts (each named in the message).
    """
    if not workers:
        raise ValueError("plan_schedule: no workers given")
    if exact_solve_limit < 1:
        raise ValueError(
            f"plan_schedule: exact_solve_limit must be >= 1, got {exact_solve_limit}"
        )
    if not durations:
        return ScheduledPlan(assignments={}, predicted_makespan_ms=0, feasible=True)

    _validate_eligibility(durations, module_infra, workers)

    if len(durations) <= exact_solve_limit:
        assignments = _exact_schedule(durations, workers, module_infra, conflicts)
    else:
        assignments = _decomposed_schedule(
            durations, workers, module_infra, conflicts, exact_solve_limit
        )

    makespan = _makespan_of(assignments, durations)
    feasible = budget_ms is None or makespan <= budget_ms
    return ScheduledPlan(
        assignments=assignments, predicted_makespan_ms=makespan, feasible=feasible
    )


def load_durations(path: str) -> dict[str, int]:
    """Load per-module durations from a JSON duration cache.

    The real, already-persisted artifact: the LPT-packing duration cache
    ``core.threading.test_executor`` maintains (default path
    ``.cache/test_durations.json``, see that module's
    ``DEFAULT_DURATION_CACHE_PATH`` / ``_save_duration_cache``) -- a flat
    JSON object of ``{module_path: seconds}`` (float), updated by
    ``run_tests_threadpool.py`` after every real run from JUnit timings.
    The raw pytest-live-observer NDJSON log (``core.threading.
    pytest_live_observer``) records only PER-TEST events with no per-module
    aggregation persisted anywhere -- this cache is the actual per-module
    artifact requested by this task's brief.

    Returns module path -> duration in whole milliseconds (rounded).

    Raises:
        OSError: the file does not exist / cannot be read.
        ValueError: the file is not a JSON object of ``{str: number}``.
    """
    with open(path, encoding="utf-8") as fh:
        data = json.load(fh)
    if not isinstance(data, dict):
        raise ValueError(
            f"{path}: expected a JSON object of {{module: seconds}}, "
            f"got {type(data).__name__}"
        )
    out: dict[str, int] = {}
    for module, seconds in data.items():
        if isinstance(seconds, bool) or not isinstance(seconds, int | float):
            raise ValueError(f"{path}: duration for {module!r} is not a number")
        if seconds < 0:
            raise ValueError(f"{path}: duration for {module!r} is negative")
        out[str(module)] = round(float(seconds) * 1000)
    return out


def load_conflicts(path: str) -> set[frozenset[str]]:
    """Load module-conflict pairs from a simple JSON list of 2-element lists.

    ``scripts/gate_parallel_safety.py`` is the intended PRODUCER of this
    file (its five gates already detect parallel-collision patterns), but
    it emits per-line ``{"file", "line", ...}`` findings today, not
    module-pair conflicts -- there is no existing conflict-pair format to
    read, so this defines the simplest reasonable one: ``[["modA",
    "modB"], ...]``. Per this task's boundary, ``gate_parallel_safety.py``
    itself is NOT modified to emit this format -- that wiring is left for a
    follow-on task.

    Raises:
        OSError: the file does not exist / cannot be read.
        ValueError: malformed JSON shape (not a list of 2-element string
            lists, or a pair naming the same module twice).
    """
    with open(path, encoding="utf-8") as fh:
        data: Any = json.load(fh)
    if not isinstance(data, list):
        raise ValueError(f"{path}: expected a JSON list of [modA, modB] pairs")
    conflicts: set[frozenset[str]] = set()
    for index, pair in enumerate(data):
        if (
            not isinstance(pair, list)
            or len(pair) != 2
            or not all(isinstance(item, str) for item in pair)
        ):
            raise ValueError(
                f"{path}: entry {index} is not a 2-element list of module "
                f"paths: {pair!r}"
            )
        module_a, module_b = pair
        if module_a == module_b:
            raise ValueError(
                f"{path}: entry {index} conflicts a module with itself: "
                f"{module_a!r}"
            )
        conflicts.add(frozenset((module_a, module_b)))
    return conflicts