scheduler.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/testing/scheduler.py
 
"""
The §20 stand-in's parallel scheduler (Plan 240; ADR-T-15 slice).
 
Purpose
-------
Turn ``--workers N | auto`` into real child processes without inventing
a second classifier. The plan-239 mechanism in ``oct.testing.parallel``
owns the resource vocabulary, the committed
``parallel_manifest.json``, the grouping and the worker cap; this
module imports all four and does exactly three things the pytest side
does not need: it resolves the requested worker count against the cap,
it turns the manifest's groups into scheduling units, and it runs those
units in a bounded pool of children, reconciling their results into the
single §11 report.
 
Responsibilities
----------------
- ``resolve_worker_plan`` — P2: ``auto`` = min(cap, cores); an explicit
  count above the cap CLAMPS and carries the note that says so.
- ``build_schedule`` — P3 as Plan 286 amended it: a group is one unit;
  a ``free`` file is cut, in collection order, into units of at most
  ``MAX_FREE_UNIT_TESTS`` tests, unless it carries the ``whole_file``
  opt-out; units run largest first by collected count; unclassified
  files share one serial unit. Deterministic for any input order.
- ``load_groups`` — the manifest read, delegated to plan 239.
- ``run_units`` — the bounded pool: one spawned child per unit, each
  with its own §20.2 snapshot, reconciled fail-closed.
- ``child_entry`` — the child-side entry point (module level so the
  ``spawn`` start method can import it).
 
Diagnostics
-----------
Domain: OC_TESTING
Levels:
    L2 — lifecycle events (pool start/stop, unit completion)
    L3 — semantic details (resolved counts, unit shapes)
    L4 — deep tracing
 
Contracts
---------
- This module NEVER classifies a test file. Every group it uses comes
  from the committed manifest through ``oct.testing.parallel``; a file
  with no entry is :data:`GROUP_UNCLASSIFIED` and is reported, never
  guessed at (P1).
- A child that dies without writing its result is a CRASH: its nodeids
  are reported failed, never passed and never dropped (P3, spec-10
  §8.4 worker-crash semantics).
- The start method is always ``spawn``, so a child inherits the
  parent's ``sys.path`` (the F17 shadow guard) and nothing else of the
  parent's in-process state.
- Ordering of everything this module returns is derived from the
  schedule, not from completion order.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oct.testing.parallel (lazy; Plan 239: manifest, groups, worker cap)
- oc_chat.testing.report (violation/error constructors)
- oc_chat.testing.runner (lazy; the in-process session, imported in the child to keep the module graph acyclic)
"""
from __future__ import annotations
 
from dataclasses import dataclass, field
import json
import multiprocessing
import os
from pathlib import Path
import tempfile
import time
from typing import Any, Iterable, Mapping, Sequence
 
from oc_diagnostics import _dbg
 
from oc_chat.testing.report import make_error, make_violation
 
#: Shipped worker cap when the workspace manifest declares none. Kept
#: in step with ``oct.testing.parallel.DEFAULT_MAX_WORKERS`` (Plan 239
#: P1, operator ruling 2026-09-06); the literal is the fallback for an
#: installed ``oct`` that predates the key.
DEFAULT_WORKER_CAP: int = 24
 
#: Group name of a file the manifest does not classify (Plan 239 P4).
GROUP_UNCLASSIFIED: str = "unclassified-serial"
 
#: Group name of a file that touches no shared resource.
GROUP_FREE: str = "free"
 
#: Marker that keeps ONE free file's tests in a single unit (Plan 286
#: P2) — the opt-out for a file whose tests depend on each other's
#: order or on module state. Mirrors
#: ``oct.testing.parallel.WHOLE_FILE_MARKER``, which the xdist path
#: reads off the collected item; here the marked PATHS arrive from the
#: parent's collect pass, because a child is handed nodeids and never
#: sees a mark.
WHOLE_FILE_MARKER: str = "whole_file"
 
#: Most tests one ``free`` scheduling unit may hold (Plan 286 P2).
#:
#: A unit is one spawned child, and a child pays a full pytest
#: start-up before it runs anything. Measured on the live oc_chat tree
#: from this worktree at BASE ``3297fae7`` (``--workers auto``, 20
#: children, 283 free units): the cheapest unit took 2.2 s wall, which
#: is that start-up, and the free set's 3877 tests took 819.6 s of
#: work beyond it — 0.21 s per test. The plan's rule is that start-up
#: stays under a TENTH of a unit's run time, so a unit must carry at
#: least 22 s of work, which is 104 tests at that rate. Hence 100: a
#: free file above it splits into chunks in collection order, and the
#: 275 free files below it stay exactly as they were, because spending
#: 2.2 s to save a fraction of a second is not parallelism.
MAX_FREE_UNIT_TESTS: int = 100
 
#: ``workers.dist_mode`` when the scheduler distributed the run
#: (spec-10 §11 value set ``{oct_det_loadscope, serial}``).
DIST_MODE_PARALLEL: str = "oct_det_loadscope"
 
#: ``workers.dist_mode`` when no scope distribution happened.
DIST_MODE_SERIAL: str = "serial"
 
#: ``9007`` — the clamp note (see the report module's catalog).
CODE_WORKERS_CLAMPED: str = "9007"
 
#: How often the pool polls its children for completion.
_POLL_SECONDS: float = 0.02
 
 
class SchedulerUnavailable(Exception):
    """Raised when the plan-239 mechanism cannot be imported."""
 
 
# ---------------------------------------------------------------------
# P1 — the plan-239 mechanism, imported and never re-implemented
# ---------------------------------------------------------------------
 
def load_parallel_module(workspace_root: Path | None = None) -> Any:
    """Import ``oct.testing.parallel``, falling back to the checkout.
 
    In a worktree the editable ``oct`` install may resolve to ANOTHER
    checkout that does not carry the module yet; the sibling
    subproject in this repository always does. Only that ONE module is
    loaded by path — the rest of ``oct`` keeps resolving normally.
    This mirrors the root conftest's fallback (Plan 239), because the
    stand-in runs in exactly the same trees.
    """
    func = "load_parallel_module"
    try:
        from oct.testing import parallel
    except ImportError:
        parallel = None
    if parallel is not None:
        return parallel
    if workspace_root is None:
        raise SchedulerUnavailable("oct.testing.parallel is not importable")
    candidate = (
        Path(workspace_root) / "oct" / "oct" / "testing" / "parallel.py"
    )
    if not candidate.is_file():
        raise SchedulerUnavailable(
            f"oct.testing.parallel is not importable and {candidate} "
            "does not exist",
        )
    import importlib.util
    import sys
 
    spec = importlib.util.spec_from_file_location(
        "oct_testing_parallel_checkout", candidate,
    )
    if spec is None or spec.loader is None:  # pragma: no cover - defensive
        raise SchedulerUnavailable(f"{candidate} is not loadable as a module")
    module = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = module
    spec.loader.exec_module(module)
    _dbg("OC_TESTING", func, f"loaded the classifier from {candidate}", 2)
    return module
 
 
def manifest_path_for(
    root: Path,
    testpaths: Sequence[str],
    workspace_root: Path | None = None,
) -> Path:
    """The committed manifest implied by *testpaths* (Plan 239 P4)."""
    parallel = load_parallel_module(workspace_root)
    roots = list(testpaths) or ["."]
    return parallel.default_manifest_path(Path(root), roots)
 
 
def load_groups(
    manifest_path: Path,
    relpaths: Iterable[str],
    workspace_root: Path | None = None,
) -> tuple[dict[str, str], tuple[str, ...]]:
    """Map root-relative files to their groups through plan 239.
 
    Returns
    -------
    tuple
        ``(path -> group, missing)``. A collected file with no manifest
        entry maps to :data:`GROUP_UNCLASSIFIED` and is listed in
        *missing* — the fail-closed state a stale manifest produces.
    """
    func = "load_groups"
    parallel = load_parallel_module(workspace_root)
    manifest = parallel.load_manifest(Path(manifest_path))
    groups, missing = parallel.resolve_groups(manifest, list(relpaths))
    _dbg("OC_TESTING", func,
         f"{len(groups)} file(s), {len(missing)} unclassified", 3)
    return dict(groups), tuple(missing)
 
 
# ---------------------------------------------------------------------
# P2 — the worker count
# ---------------------------------------------------------------------
 
@dataclass(frozen=True)
class WorkerPlan:
    """The resolved ``--workers`` decision.
 
    Attributes
    ----------
    requested
        The CLI value verbatim: a positive integer or ``"auto"``.
    cap
        ``subprojects[].test.max_workers`` for this package, or the
        shipped default when the workspace manifest declares none.
    resolved
        The pool size this run may use.
    clamped
        True when an explicit request exceeded *cap*.
    clamp_message
        The human note naming both numbers; ``None`` when not clamped.
    """
 
    requested: int | str
    cap: int
    resolved: int
    clamped: bool = False
    clamp_message: str | None = None
 
 
def resolve_worker_plan(
    requested: int | str,
    workspace_root: Path,
    package: str,
    *,
    cpu_count: int | None = None,
) -> WorkerPlan:
    """Resolve *requested* against the plan-239 cap (P2).
 
    ``auto`` resolves to ``min(cap, cores)``; an explicit count above
    the cap is clamped to it and the clamp is reported, never silent.
    """
    func = "resolve_worker_plan"
    parallel = load_parallel_module(Path(workspace_root))
    try:
        cap = int(parallel.resolve_max_workers(Path(workspace_root), package))
    except Exception as exc:  # EXTERNAL_ERROR(L2) - a malformed manifest
        # Plan 239 P1 refuses rather than degrades on an invalid
        # ``max_workers``. The stand-in keeps that invariant the only way
        # it safely can: it declines to parallelise at all, which is more
        # serialisation and never less.
        _dbg("OC_TESTING", func, f"cap unusable: {exc}", 1)
        raise SchedulerUnavailable(
            f"the workspace manifest's worker cap for {package!r} is "
            f"unusable: {exc}",
        ) from exc
    cap = max(1, cap)
    cores = cpu_count if cpu_count is not None else (os.cpu_count() or 1)
    cores = max(1, cores)
 
    if requested == "auto":
        resolved = max(1, min(cap, cores))
        _dbg("OC_TESTING", func, f"auto -> {resolved} (cap {cap}, "
             f"{cores} core(s))", 3)
        return WorkerPlan(requested="auto", cap=cap, resolved=resolved)
 
    count = max(1, int(requested))
    if count > cap:
        message = (
            f"--workers {count} exceeds the manifest cap for {package!r} "
            # OCT-LINT: disable=no_hardcoded_secrets reason="the dotted workspace-manifest key path is prose telling an operator which key to edit; it is not a credential"
            f"and was clamped to {cap} "
            "(subprojects[].test.max_workers, Plan 239 P1)"
        )
        _dbg("OC_TESTING", func, message, 2)
        return WorkerPlan(requested=count, cap=cap, resolved=cap,
                          clamped=True, clamp_message=message)
    return WorkerPlan(requested=count, cap=cap, resolved=count)
 
 
def clamp_error(plan: WorkerPlan) -> dict:
    """The §11 ``errors[]`` entry that reports a clamp."""
    return make_error(
        "configuration", CODE_WORKERS_CLAMPED, "warning",
        plan.clamp_message or "",
    )
 
 
# ---------------------------------------------------------------------
# P3 — groups become scheduling units
# ---------------------------------------------------------------------
 
@dataclass(frozen=True)
class ScheduleUnit:
    """One unit of scheduling: everything one child will run.
 
    Attributes
    ----------
    group
        The plan-239 group name this unit carries.
    worker_id
        ``gw<n>``, assigned by the unit's position in the deterministic
        schedule, so every nodeid of a group shares one worker id
        regardless of which OS process picked it up.
    files
        Root-relative POSIX paths, sorted.
    nodeids
        The parent's selection for those files, in collection order.
    """
 
    group: str
    worker_id: str
    files: tuple[str, ...]
    nodeids: tuple[str, ...]
 
 
def build_schedule(
    groups: Mapping[str, str],
    nodeids_by_file: Mapping[str, Sequence[str]],
    *,
    whole_file: Iterable[str] = (),
    max_free_unit_tests: int | None = None,
) -> tuple[ScheduleUnit, ...]:
    """Turn the plan-239 grouping into scheduling units (P3, Plan 286).
 
    A group is the unit of scheduling: every file of a group runs in
    ONE child, because that is what the group means, and the
    unclassified set shares one serial unit because fail-closed means
    serial.
 
    A ``free`` file shares no resource with anything, so its tests may
    run anywhere — including in different children (Plan 286 P2). Its
    nodeids are therefore cut, in collection order, into units of at
    most *max_free_unit_tests* tests, so that one file with hundreds of
    tests no longer holds a child for its whole length. A file listed
    in *whole_file* is the opt-out: it stays ONE unit however many
    tests it has.
 
    Units are scheduled largest first by collected count (longest
    processing time first — the classic bound for a bounded pool) and
    the result is deterministic for any input order.
 
    Parameters
    ----------
    groups
        Root-relative path -> plan-239 group name.
    nodeids_by_file
        Root-relative path -> the parent's selection, in collection
        order.
    whole_file
        Root-relative paths whose tests must stay in one unit: the
        files the parent's collect pass saw carrying
        :data:`WHOLE_FILE_MARKER`.
    max_free_unit_tests
        Override for :data:`MAX_FREE_UNIT_TESTS`; values below 1 are
        read as 1, so a caller can never ask for an empty unit.
    """
    func = "build_schedule"
    limit = (
        MAX_FREE_UNIT_TESTS if max_free_unit_tests is None
        else max(1, int(max_free_unit_tests))
    )
    intact = {str(path) for path in whole_file}
    # key -> the files it holds; one entry per unit, in a stable order.
    buckets: dict[str, list[str]] = {}
    # key -> the unit's own nodeids, for a free file cut into chunks.
    chunks: dict[str, tuple[str, ...]] = {}
    for path in sorted(nodeids_by_file):
        group = groups.get(path, GROUP_UNCLASSIFIED)
        if group != GROUP_FREE:
            buckets.setdefault(group, []).append(path)
            continue
        nodeids = tuple(nodeids_by_file.get(path, ()))
        span = len(nodeids) if path in intact else limit
        for start in range(0, max(1, len(nodeids)), max(1, span)):
            key = f"{path}#{start // max(1, span):04d}"
            buckets.setdefault(key, []).append(path)
            chunks[key] = nodeids[start:start + span]
 
    def _tests(key: str, paths: Sequence[str]) -> int:
        if key in chunks:
            return len(chunks[key])
        return sum(len(nodeids_by_file.get(path, ())) for path in paths)
 
    ordered = sorted(
        buckets.items(),
        key=lambda item: (-_tests(item[0], item[1]), item[0]),
    )
    units: list[ScheduleUnit] = []
    for index, (key, paths) in enumerate(ordered):
        files = tuple(sorted(paths))
        nodeids = chunks.get(key) or tuple(
            nodeid for path in files
            for nodeid in nodeids_by_file.get(path, ())
        )
        if not nodeids:
            continue
        units.append(ScheduleUnit(
            group=groups.get(files[0], GROUP_UNCLASSIFIED),
            worker_id=f"gw{index}",
            files=files,
            nodeids=nodeids,
        ))
    _dbg("OC_TESTING", func, f"{len(units)} unit(s) from "
         f"{len(nodeids_by_file)} file(s)", 3)
    return tuple(units)
 
 
# ---------------------------------------------------------------------
# The child protocol
# ---------------------------------------------------------------------
 
@dataclass(frozen=True)
class ChildRequest:
    """Everything a child needs; picklable by construction."""
 
    root: str
    pytest_args: tuple[str, ...]
    nodeids: tuple[str, ...]
    group: str
    worker_id: str
    timeout: int | None
 
 
@dataclass
class UnitOutcome:
    """One unit's reconciled result.
 
    ``crashed`` is the fail-closed state: the child died without
    writing a result, so nothing it may have reported can be trusted.
    """
 
    unit: ScheduleUnit
    records: dict[str, dict] = field(default_factory=dict)
    timed_out: tuple[str, ...] = ()
    collect_errors: tuple[str, ...] = ()
    violations: list[dict] = field(default_factory=list)
    rc: int = 0
    pid: int | None = None
    started_seconds: float = 0.0
    finished_seconds: float = 0.0
    crashed: bool = False
    crash_detail: str = ""
    output: str = ""
 
    @property
    def duration_seconds(self) -> float:
        """Wall clock the unit occupied, in seconds."""
        return max(0.0, self.finished_seconds - self.started_seconds)
 
 
def child_entry(request: ChildRequest, result_path: str) -> None:
    """Run one unit and write its result document (the spawn target).
 
    Imports the runner lazily: the runner imports THIS module at module
    scope, and a child that re-imported it eagerly would close the
    cycle. Any exception is recorded in the result document rather than
    raised, so the parent distinguishes "the unit failed" from "the
    child died".
    """
    func = "child_entry"
    payload: dict[str, Any]
    try:
        from oc_chat.testing.runner import run_session
 
        outcome = run_session(
            pytest_args=[*request.pytest_args, *request.nodeids],
            root=Path(request.root),
            timeout=request.timeout,
            dry_run=False,
            install_signal_handlers=False,
            violation_file=request.root,
            violation_scope=f"{request.worker_id}/{request.group}",
        )
        payload = {
            "rc": outcome.rc,
            "records": outcome.records,
            "timed_out": sorted(outcome.timed_out),
            "collect_errors": list(outcome.collect_errors),
            "violations": outcome.violations,
            "pid": os.getpid(),
            "output": outcome.output,
            "internal_error": None,
        }
    except BaseException as exc:  # noqa: BLE001 - reported, never raised
        _dbg("OC_TESTING", func, f"child internal error: {exc!r}", 1)
        payload = {
            "rc": 3,
            "records": {},
            "timed_out": [],
            "collect_errors": [],
            "violations": [],
            "pid": os.getpid(),
            "output": "",
            "internal_error": f"{type(exc).__name__}: {exc}"[:2000],
        }
    Path(result_path).write_text(
        json.dumps(payload), encoding="utf-8", newline="\n",
    )
 
 
# ---------------------------------------------------------------------
# The bounded pool
# ---------------------------------------------------------------------
 
def _spawn(
    unit: ScheduleUnit,
    *,
    root: Path,
    pytest_args: Sequence[str],
    timeout: int | None,
    result_path: Path,
) -> Any:
    """Start one child for *unit* and return the live process."""
    context = multiprocessing.get_context("spawn")
    request = ChildRequest(
        root=str(root),
        pytest_args=tuple(pytest_args),
        nodeids=unit.nodeids,
        group=unit.group,
        worker_id=unit.worker_id,
        timeout=timeout,
    )
    process = context.Process(
        target=child_entry,
        args=(request, str(result_path)),
        name=f"oc-standin-{unit.worker_id}",
        daemon=False,
    )
    process.start()
    return process
 
 
def _read_result(
    unit: ScheduleUnit,
    process: Any,
    result_path: Path,
    started: float,
    finished: float,
) -> UnitOutcome:
    """Turn a finished child into a :class:`UnitOutcome`, fail-closed."""
    func = "_read_result"
    outcome = UnitOutcome(
        unit=unit, pid=process.pid,
        started_seconds=started, finished_seconds=finished,
    )
    try:
        payload = json.loads(result_path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        outcome.crashed = True
        outcome.crash_detail = (
            f"exit status {process.exitcode}; no readable result document "
            f"({type(exc).__name__})"
        )
        _dbg("OC_TESTING", func,
             f"{unit.worker_id} crashed: {outcome.crash_detail}", 1)
        return outcome
    if payload.get("internal_error"):
        outcome.crashed = True
        outcome.crash_detail = str(payload["internal_error"])
        outcome.pid = payload.get("pid", outcome.pid)
        return outcome
    outcome.records = dict(payload.get("records") or {})
    outcome.timed_out = tuple(payload.get("timed_out") or ())
    outcome.collect_errors = tuple(payload.get("collect_errors") or ())
    outcome.violations = list(payload.get("violations") or [])
    outcome.rc = int(payload.get("rc", 0))
    outcome.pid = payload.get("pid", outcome.pid)
    outcome.output = str(payload.get("output") or "")
    return outcome
 
 
def run_units(
    units: Sequence[ScheduleUnit],
    *,
    workers: int,
    root: Path,
    pytest_args: Sequence[str],
    timeout: int | None,
    origin: float,
) -> tuple[list[UnitOutcome], bool]:
    """Run *units* in a bounded pool of children.
 
    Parameters
    ----------
    workers
        Maximum number of children alive at once.
    origin
        The run's ``time.monotonic()`` zero, so every window this
        returns is relative to the report's own clock.
 
    Returns
    -------
    tuple
        ``(outcomes in schedule order, interrupted)``. On interruption
        the still-running children are terminated and the units that
        never started are absent — the caller reports a partial run.
    """
    func = "run_units"
    _dbg("OC_TESTING", func,
         f"{len(units)} unit(s) over {workers} worker(s)", 2)
    pending = list(units)
    running: list[tuple[ScheduleUnit, Any, Path, float]] = []
    done: dict[str, UnitOutcome] = {}
    interrupted = False
 
    with tempfile.TemporaryDirectory(prefix="oc_standin_") as scratch:
        scratch_dir = Path(scratch)
        try:
            while pending or running:
                while pending and len(running) < workers:
                    unit = pending.pop(0)
                    result_path = scratch_dir / f"{unit.worker_id}.json"
                    started = time.monotonic() - origin
                    process = _spawn(
                        unit, root=root, pytest_args=pytest_args,
                        timeout=timeout, result_path=result_path,
                    )
                    running.append((unit, process, result_path, started))
                time.sleep(_POLL_SECONDS)
                for entry in list(running):
                    unit, process, result_path, started = entry
                    if process.is_alive():
                        continue
                    process.join()
                    finished = time.monotonic() - origin
                    done[unit.worker_id] = _read_result(
                        unit, process, result_path, started, finished,
                    )
                    running.remove(entry)
        except KeyboardInterrupt:
            interrupted = True
            _dbg("OC_TESTING", func, "interrupted; terminating children", 1)
            for _unit, process, _path, _started in running:
                if process.is_alive():
                    process.terminate()
            for _unit, process, _path, _started in running:
                process.join(timeout=10)
 
    ordered = [done[unit.worker_id] for unit in units
               if unit.worker_id in done]
    _dbg("OC_TESTING", func,
         f"{len(ordered)} unit(s) reconciled, interrupted={interrupted}", 2)
    return ordered, interrupted
 
 
def crash_violation(outcome: UnitOutcome) -> dict:
    """The §11 ``contract_violations[]`` row a dead child produces."""
    return make_violation(
        "child_crash", outcome.unit.files[0], None, "error",
        f"scheduler child {outcome.unit.worker_id} running group "
        f"{outcome.unit.group!r} died before reporting "
        f"({outcome.crash_detail}); its "
        f"{len(outcome.unit.nodeids)} nodeid(s) are reported failed, "
        "never passed (Plan 240 P3, spec-10 section 8.4)",
    )
 
 
def crash_error(outcome: UnitOutcome) -> dict:
    """The §12.4 ``WORKER_CRASHED`` entry a dead child produces."""
    return make_error(
        "orchestrator", "WORKER_CRASHED", "warning",
        f"worker {outcome.unit.worker_id} (group {outcome.unit.group!r}) "
        f"crashed: {outcome.crash_detail}",
    )
 
 
def unclassified_violation(root: Path, missing: Sequence[str]) -> dict:
    """The §11 row an absent manifest entry produces (P1)."""
    listed = ", ".join(sorted(missing))
    return make_violation(
        "unclassified_test_file", str(root), None, "error",
        f"{len(missing)} collected file(s) carry no parallel_manifest.json "
        f"entry and were run serially rather than in parallel: {listed}. "
        "Regenerate the manifest (python -m oct.testing.parallel classify) "
        "— a stale manifest may never silently widen parallelism",
    )
 
 
_dbg("OC_TESTING", "module", "loaded", 4)
 
__all__ = [
    "CODE_WORKERS_CLAMPED",
    "DEFAULT_WORKER_CAP",
    "DIST_MODE_PARALLEL",
    "DIST_MODE_SERIAL",
    "GROUP_FREE",
    "GROUP_UNCLASSIFIED",
    "MAX_FREE_UNIT_TESTS",
    "WHOLE_FILE_MARKER",
    "ChildRequest",
    "ScheduleUnit",
    "SchedulerUnavailable",
    "UnitOutcome",
    "WorkerPlan",
    "build_schedule",
    "child_entry",
    "clamp_error",
    "crash_error",
    "crash_violation",
    "load_groups",
    "load_parallel_module",
    "manifest_path_for",
    "resolve_worker_plan",
    "run_units",
    "unclassified_violation",
]