Hands-on lab — Day 143: The Machine Learning Workflow

#!/usr/bin/env python3
"""Print every measured pair in this lab as one table.

The harness compares this output byte for byte against
expected-output/measured-values.txt, so the report is not a convenience:
it is how the lab notices that a number in the lesson has gone stale.
"""

import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))

import workflow_lib as w  # noqa: E402


def rule(title: str) -> None:
    print()
    print(title)
    print("-" * len(title))


def main() -> None:
    print("Day 143 -- the machine learning workflow, measured")
    print("=" * 49)

    rule("1. The pipeline as stages, with a step log")
    honest = w.run_pipeline(w.honest_stages(), w.starting_artifact())
    for name, produced in honest.log:
        print(f"    {name:<14} -> {', '.join(produced)}")

    rule("2. The honest pipeline on data that is pure noise")
    print(f"  cross-validated accuracy               : {honest.data['score']:.4f}")
    print(f"  majority-class baseline                : {honest.data['baseline']:.4f}")
    print(f"  per-fold scores                        : {honest.data['fold_scores'].tolist()}")
    print("  the labels are coin flips, so chance is the only honest answer")

    rule("3. The same five stages, two of them transposed")
    leaky = w.run_pipeline(w.leaky_stages(), w.starting_artifact(), enforce_contracts=False)
    print(f"  split then select (correct)            : {honest.data['score']:.4f}")
    print(f"  select then split (wrong, unchecked)   : {leaky.data['score']:.4f}")
    print(f"  accuracy invented by the reordering    : {leaky.data['score'] - honest.data['score']:.4f}")
    print(f"    correct order : {' -> '.join(n for n, _ in honest.log)}")
    print(f"    wrong order   : {' -> '.join(n for n, _ in leaky.log)}")
    try:
        w.run_pipeline(w.leaky_stages(), w.starting_artifact(), enforce_contracts=True)
        print("  with contracts enforced                : NO ERROR RAISED")
    except w.StageContractError as exc:
        print(f"  with contracts enforced                : {type(exc).__name__}")
        print(f"    {exc}")

    rule("3b. How the inflation grows with the number of features chosen")
    X, y = w.noise_dataset()
    print("    k    wrong   right   invented")
    for k, wrong, right, gap in w.inflation_by_k(X, y, [5, 10, 20, 50]):
        print(f"    {k:2d}   {wrong:.4f}  {right:.4f}   {gap:+.4f}")

    rule("4. The metric decides which model you ship")
    train = w.imbalanced_dataset(1000, 11)
    test = w.imbalanced_dataset(2000, 12)
    scores = w.score_all(w.candidate_models(), train[0], train[1], test[0], test[1])
    print(f"  {'model':<30}{'acc':>8}{'prec':>8}{'recall':>8}{'f1':>8}")
    for name, s in scores.items():
        print(
            f"  {name:<30}{s['accuracy']:>8.4f}{s['precision']:>8.4f}"
            f"{s['recall']:>8.4f}{s['f1']:>8.4f}"
        )
    for metric in ("accuracy", "precision", "recall", "f1"):
        print(f"  best by {metric:<10}: {w.winner(scores, metric)}")
    print(f"  test set carries {int(test[1].sum())} positives in {len(test[1])} rows")

    rule("5. Error analysis: what the accuracy hides")
    model = w.candidate_models()["logistic (default threshold)"]
    model.fit(train[0], train[1])
    table = w.error_table(model, test[0], test[1])
    print("  rows = true class, columns = predicted class")
    print(f"    true 0: {table[0]}")
    print(f"    true 1: {table[1]}")
    print(f"  94.35 percent accurate, and it misses {table[1][0]} positives while catching {table[1][1]}")

    rule("6. Reproducibility: the same inputs give the same artifact")
    keys = ("X", "y", "fold_scores", "score")
    first = w.manifest(w.run_pipeline(w.honest_stages(), w.starting_artifact()), keys)
    second = w.manifest(w.run_pipeline(w.honest_stages(), w.starting_artifact()), keys)
    other = w.manifest(w.run_pipeline(w.honest_stages(), w.starting_artifact(seed=144)), keys)
    for key in sorted(first):
        print(f"    {key:<12} {first[key]}")
    print(f"  two runs at seed 143 agree             : {first == second}")
    print(f"  a run at seed 144 differs              : {first != other}")

    rule("7. The modelling stage is the small one")
    lines = w.stage_source_lines(w.honest_stages())
    total = sum(lines.values())
    for name in ("load", "split", "select", "fit_and_score", "baseline"):
        print(f"    {name:<14} {lines[name]:>3} lines")
    print(f"    {'total':<14} {total:>3} lines")
    print(f"  the fitting stage is {lines['fit_and_score'] / total:.2%} of this pipeline")
    print("  and this pipeline has no cleaning, monitoring or deployment stage,")
    print("  so that figure is an upper bound rather than an estimate")


if __name__ == "__main__":
    main()

"""The reference solutions: what the workflow is, and what its order costs.

Every number here was captured from a real run of this file on the
authoring machine. If a number changes, the claim in the lesson is wrong
and one of the two must be fixed.
"""

import numpy as np
import pytest

import workflow_lib as w


@pytest.fixture(scope="module")
def noise():
    return w.noise_dataset()


@pytest.fixture(scope="module")
def imbalanced():
    return w.imbalanced_dataset(1000, 11), w.imbalanced_dataset(2000, 12)


# --- 1. The workflow is stages with contracts, not boxes with arrows ------


def test_01_the_honest_pipeline_runs_in_the_declared_order():
    result = w.run_pipeline(w.honest_stages(), w.starting_artifact())
    assert [name for name, _keys in result.log] == [
        "load",
        "split",
        "select",
        "fit_and_score",
        "baseline",
    ]
    # Each stage recorded exactly the keys it declared it would produce.
    produced = dict(result.log)
    assert produced["load"] == ("X", "y")
    assert produced["split"] == ("folds",)
    assert produced["select"] == ("selected",)
    assert produced["fit_and_score"] == ("fold_scores", "score")
    assert produced["baseline"] == ("baseline",)


def test_01b_a_stage_never_mutates_the_artifact_it_was_given():
    start = w.starting_artifact()
    before = set(start.data)
    w.run_pipeline(w.honest_stages(), start)
    # The starting artifact is untouched: every stage returned a new one.
    assert set(start.data) == before
    assert start.log == []


# --- 2. The honest pipeline reports chance on data that is chance --------


def test_02_the_honest_pipeline_reports_chance_on_pure_noise():
    result = w.run_pipeline(w.honest_stages(), w.starting_artifact())
    assert result.data["score"] == 0.5
    assert result.data["baseline"] == 0.54
    assert result.data["fold_scores"].shape == (5,)
    # Labels are coin flips, so chance is the only honest answer.
    assert abs(result.data["score"] - 0.5) < 0.01


# --- 3. The same stages in the wrong order, and what a contract is for ---


def test_03_reordering_two_stages_invents_twenty_three_accuracy_points():
    honest = w.run_pipeline(w.honest_stages(), w.starting_artifact())
    leaky = w.run_pipeline(w.leaky_stages(), w.starting_artifact(), enforce_contracts=False)
    assert honest.data["score"] == 0.5
    assert leaky.data["score"] == 0.73
    assert round(leaky.data["score"] - honest.data["score"], 4) == 0.23
    # The step logs differ by one transposition and nothing else.
    assert [n for n, _ in honest.log] == ["load", "split", "select", "fit_and_score", "baseline"]
    assert [n for n, _ in leaky.log] == ["load", "select", "split", "fit_and_score", "baseline"]


def test_03b_the_contract_turns_a_silent_lie_into_a_named_failure():
    with pytest.raises(w.StageContractError) as excinfo:
        w.run_pipeline(w.leaky_stages(), w.starting_artifact(), enforce_contracts=True)
    message = str(excinfo.value)
    assert "'select'" in message
    assert "folds" in message
    # The honest pipeline passes the same contracts without complaint.
    ok = w.run_pipeline(w.honest_stages(), w.starting_artifact(), enforce_contracts=True)
    assert ok.data["score"] == 0.5


def test_03c_the_inflation_grows_with_the_number_of_features_chosen(noise):
    X, y = noise
    rows = w.inflation_by_k(X, y, [5, 10, 20, 50])
    assert rows == [
        (5, 0.65, 0.39, 0.26),
        (10, 0.72, 0.5, 0.22),
        (20, 0.73, 0.5, 0.23),
        (50, 0.85, 0.38, 0.47),
    ]
    # Every wrong-order score beats its honest counterpart, at every k.
    assert all(wrong > right for _k, wrong, right, _gap in rows)
    # And the honest scores stay at or below chance, as they must.
    assert all(right <= 0.5 for _k, _wrong, right, _gap in rows)


# --- 4. The metric is chosen before the model, and it decides ------------


def test_04_the_metric_you_choose_decides_which_model_you_ship(imbalanced):
    (X_train, y_train), (X_test, y_test) = imbalanced
    scores = w.score_all(w.candidate_models(), X_train, y_train, X_test, y_test)
    assert scores["majority baseline"] == {
        "accuracy": 0.92,
        "precision": 0.0,
        "recall": 0.0,
        "f1": 0.0,
    }
    assert scores["logistic (default threshold)"]["accuracy"] == 0.9435
    assert scores["logistic (default threshold)"]["recall"] == 0.4813
    assert scores["logistic (balanced)"]["accuracy"] == 0.8685
    assert scores["logistic (balanced)"]["recall"] == 0.8438
    # The decision inverts on the metric, with nothing else changing.
    assert w.winner(scores, "accuracy") == "logistic (default threshold)"
    assert w.winner(scores, "recall") == "logistic (balanced)"
    assert w.winner(scores, "accuracy") != w.winner(scores, "recall")


def test_04b_a_model_that_never_predicts_the_positive_class_scores_ninety_two(
    imbalanced,
):
    (X_train, y_train), (X_test, y_test) = imbalanced
    scores = w.score_all(w.candidate_models(), X_train, y_train, X_test, y_test)
    baseline = scores["majority baseline"]
    assert baseline["accuracy"] == 0.92 and baseline["recall"] == 0.0
    beats = {
        name: s["accuracy"]
        for name, s in scores.items()
        if s["accuracy"] > baseline["accuracy"]
    }
    # Three of the four real models beat a constant -- by at most 2.35 points.
    assert set(beats) == {"logistic (default threshold)", "5-NN", "depth-3 tree"}
    assert round(max(beats.values()) - baseline["accuracy"], 4) == 0.0235
    # And the one that actually finds positives is the one that loses here.
    assert scores["logistic (balanced)"]["accuracy"] < baseline["accuracy"]
    assert scores["logistic (balanced)"]["recall"] == 0.8438
    assert int(y_test.sum()) == 160 and len(y_test) == 2000


# --- 5. Error analysis is a stage, not an afterthought ------------------


def test_05_the_confusion_matrix_says_what_the_accuracy_hides(imbalanced):
    (X_train, y_train), (X_test, y_test) = imbalanced
    model = w.candidate_models()["logistic (default threshold)"]
    model.fit(X_train, y_train)
    table = w.error_table(model, X_test, y_test)
    assert table == [[1810, 30], [83, 77]]
    true_negative, false_positive = table[0]
    false_negative, true_positive = table[1]
    assert true_negative + false_positive + false_negative + true_positive == 2000
    # 94.35% accurate, and it misses more positives than it catches.
    assert false_negative > true_positive
    assert round((true_negative + true_positive) / 2000, 4) == 0.9435


# --- 6. Reproducibility: the same inputs must give the same artifact ----


def test_06_two_runs_of_the_pipeline_are_byte_identical():
    keys = ("X", "y", "fold_scores", "score")
    first = w.manifest(w.run_pipeline(w.honest_stages(), w.starting_artifact()), keys)
    second = w.manifest(w.run_pipeline(w.honest_stages(), w.starting_artifact()), keys)
    assert first == second
    assert first == {
        "X": "51b0a421bd652dd2",
        "fold_scores": "8f0ac332958b9bc4",
        "score": "d2cbad71ff333de6",
        "y": "9984503b5352c5a1",
    }


def test_06b_a_different_seed_produces_a_different_manifest():
    keys = ("X", "y", "fold_scores", "score")
    first = w.manifest(w.run_pipeline(w.honest_stages(), w.starting_artifact()), keys)
    other = w.manifest(
        w.run_pipeline(w.honest_stages(), w.starting_artifact(seed=144)), keys
    )
    assert first != other
    # A manifest that never changes is not evidence of determinism.
    assert first["X"] != other["X"]


# --- 7. The modelling stage is the small one ---------------------------


def test_07_the_modelling_stage_is_the_smallest_part_of_the_pipeline():
    lines = w.stage_source_lines(w.honest_stages())
    assert lines == {"load": 5, "split": 2, "select": 10, "fit_and_score": 9, "baseline": 4}
    total = sum(lines.values())
    assert total == 30
    assert round(lines["fit_and_score"] / total, 4) == 0.3
    # And this pipeline has no cleaning, no monitoring and no deployment
    # stage at all -- so 30% is an upper bound, not an estimate.
    assert set(lines) == {"load", "split", "select", "fit_and_score", "baseline"}
    assert "clean" not in lines and "monitor" not in lines and "deploy" not in lines


# --- 8. A missing input is caught before it becomes a wrong number ------


def test_08_a_stage_cannot_run_without_the_inputs_it_declared():
    stages = w.honest_stages()
    empty = w.Artifact(data={})
    with pytest.raises(w.StageContractError) as excinfo:
        w.run_pipeline(stages, empty)
    assert "'load'" in str(excinfo.value)
    # With contracts off it fails too -- but as a KeyError deep inside the
    # stage, naming a dictionary key rather than the stage that broke.
    with pytest.raises(KeyError):
        w.run_pipeline(stages, empty, enforce_contracts=False)

"""Machinery checks: the runner itself behaves, before any claim is made.

These four tests are solved in both `starter/` and `examples/`. They exist
so that a broken runner reports itself as a broken runner rather than as a
surprising scientific result.
"""

import numpy as np
import pytest

import workflow_lib as w


def test_the_artifact_is_immutable_and_carries_its_history():
    start = w.Artifact(data={"a": 1})
    nxt = start.with_(b=2)
    assert start.data == {"a": 1}
    assert nxt.data == {"a": 1, "b": 2}
    assert nxt is not start
    # Overwriting a key is allowed; silently mutating the original is not.
    third = nxt.with_(a=99)
    assert third.data["a"] == 99 and nxt.data["a"] == 1


def test_the_fingerprint_is_stable_content_addressing():
    a = np.arange(10)
    b = np.arange(10)
    assert w.fingerprint(a) == w.fingerprint(b)
    assert w.fingerprint(a) != w.fingerprint(np.arange(11))
    # dtype is part of the identity: the same values in a different type
    # are a different artifact, and treating them as equal hides real bugs.
    assert w.fingerprint(np.arange(10)) != w.fingerprint(np.arange(10, dtype=float))
    assert len(w.fingerprint(a)) == 16


def test_the_contract_checks_both_directions():
    called = []

    def under_producing(_artifact):
        called.append("under")
        return {}

    stages = [w.Stage("under", under_producing, requires=(), produces=("x",))]
    with pytest.raises(w.StageContractError) as excinfo:
        w.run_pipeline(stages, w.Artifact(data={}))
    assert "'under'" in str(excinfo.value)
    assert called == ["under"]

    def over_producing(_artifact):
        return {"x": 1, "surprise": 2}

    stages = [w.Stage("over", over_producing, requires=(), produces=("x",))]
    with pytest.raises(w.StageContractError):
        w.run_pipeline(stages, w.Artifact(data={}))


def test_the_folds_are_stratified_and_cover_every_row_exactly_once():
    X, y = w.imbalanced_dataset(500, seed=3)
    splits = w.folds(X, y, n_splits=5, seed=143)
    assert len(splits) == 5
    seen = np.concatenate([test for _train, test in splits])
    assert sorted(seen.tolist()) == list(range(500))
    # Stratified: every fold carries roughly the population positive rate.
    rate = float(y.mean())
    for _train, test in splits:
        assert abs(float(y[test].mean()) - rate) < 0.02

"""The machine learning workflow as runnable stages, with contracts.

The workflow in the textbook is a row of boxes with arrows. This module is
the same workflow with the arrows made load-bearing: every stage declares
what it requires and what it produces, the runner refuses to run a stage
whose inputs are absent, and every run leaves a step log and a manifest of
content hashes behind it.

That is not ceremony. The point of this lab is that **the same three
stages in a different order produce a different number**, and that the
wrong order is silent. A stage contract is what turns a silent
twenty-three point lie into a loud error naming the stage that broke.
"""

from __future__ import annotations

import hashlib
import inspect
from dataclasses import dataclass, field
from typing import Callable

import numpy as np

from sklearn.dummy import DummyClassifier
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import (
    accuracy_score,
    confusion_matrix,
    f1_score,
    precision_score,
    recall_score,
)
from sklearn.model_selection import StratifiedKFold
from sklearn.neighbors import KNeighborsClassifier
from sklearn.tree import DecisionTreeClassifier


class StageContractError(RuntimeError):
    """Raised when a stage is asked to run without the inputs it declared."""


# --------------------------------------------------------------------------
# The runner
# --------------------------------------------------------------------------


@dataclass
class Artifact:
    """Everything the pipeline knows so far, plus how it came to know it."""

    data: dict = field(default_factory=dict)
    log: list = field(default_factory=list)

    def with_(self, **produced) -> "Artifact":
        """Return a new artifact carrying the additional keys.

        Deliberately not in-place. A stage that mutates its input makes the
        step log a work of fiction, because the log then describes states
        that no longer exist.
        """
        merged = dict(self.data)
        merged.update(produced)
        return Artifact(data=merged, log=list(self.log))


@dataclass
class Stage:
    """One step of the workflow, with its input and output contract."""

    name: str
    run: Callable[[Artifact], dict]
    requires: tuple = ()
    produces: tuple = ()


def run_pipeline(stages, artifact: Artifact, enforce_contracts: bool = True) -> Artifact:
    """Run every stage in order, recording what each one did.

    With ``enforce_contracts=False`` the runner behaves like most real
    pipelines: it runs whatever it is given, in whatever order, and reports
    a number. Exercise 3 measures what that costs.
    """
    for stage in stages:
        if enforce_contracts:
            missing = [k for k in stage.requires if k not in artifact.data]
            if missing:
                raise StageContractError(
                    f"stage {stage.name!r} requires {missing} which no earlier stage produced"
                )
        produced = stage.run(artifact)
        if enforce_contracts:
            unexpected = sorted(set(produced) - set(stage.produces))
            absent = sorted(set(stage.produces) - set(produced))
            if absent or unexpected:
                raise StageContractError(
                    f"stage {stage.name!r} declared {list(stage.produces)} "
                    f"but produced {sorted(produced)}"
                )
        artifact = artifact.with_(**produced)
        artifact.log.append((stage.name, tuple(sorted(produced))))
    return artifact


def step_log(artifact: Artifact):
    """The ordered record of which stage produced which keys."""
    return list(artifact.log)


def fingerprint(value) -> str:
    """A stable content hash for an array, a number or a string."""
    if isinstance(value, np.ndarray):
        payload = np.ascontiguousarray(value).tobytes() + str(value.dtype).encode()
    else:
        payload = repr(value).encode()
    return hashlib.sha256(payload).hexdigest()[:16]


def manifest(artifact: Artifact, keys) -> dict:
    """Content hashes for the named keys, so two runs can be compared.

    A pipeline that cannot prove it produced the same thing twice cannot be
    debugged, because you can never tell a fix from a coincidence.
    """
    return {k: fingerprint(artifact.data[k]) for k in sorted(keys)}


def stage_source_lines(stages) -> dict:
    """How many lines of code each stage's function actually is.

    Self-referential on purpose: exercise 7 uses this to measure the shape
    of this very pipeline, rather than repeating the folklore figure about
    how much of the work is not modelling.
    """
    out = {}
    for stage in stages:
        source = inspect.getsource(stage.run)
        lines = [ln for ln in source.splitlines() if ln.strip() and not ln.strip().startswith("#")]
        out[stage.name] = len(lines)
    return out


# --------------------------------------------------------------------------
# Datasets
# --------------------------------------------------------------------------


def noise_dataset(n_samples: int = 100, n_features: int = 5000, seed: int = 143):
    """Labels are coin flips. No feature carries any information whatsoever.

    Any workflow that reports better than chance on this data has a bug,
    and the bug is what exercise 3 is about.
    """
    rng = np.random.default_rng(seed)
    X = rng.normal(size=(n_samples, n_features))
    y = rng.integers(0, 2, size=n_samples)
    return X, y


def imbalanced_dataset(n: int, seed: int, rate: float = 0.08):
    """A rare positive class whose features overlap the negatives.

    Eight percent positive, shifted by 1.1 standard deviations. Separable
    enough to be worth modelling, overlapping enough that the metric you
    choose decides which model wins.
    """
    rng = np.random.default_rng(seed)
    n_pos = int(round(n * rate))
    y = np.zeros(n, dtype=int)
    y[:n_pos] = 1
    X = rng.normal(size=(n, 4))
    X[y == 1] += 1.1
    order = rng.permutation(n)
    return X[order], y[order]


# --------------------------------------------------------------------------
# The individual steps, usable on their own
# --------------------------------------------------------------------------


def correlation_ranking(X, y):
    """Absolute correlation between each column and the label."""
    y = np.asarray(y, dtype=float)
    out = np.empty(X.shape[1])
    for j in range(X.shape[1]):
        out[j] = abs(float(np.corrcoef(X[:, j], y)[0, 1]))
    return out


def top_k_features(X, y, k: int):
    """The k columns most correlated with the label, as indices."""
    return np.argsort(correlation_ranking(X, y))[-k:]


def folds(X, y, n_splits: int = 5, seed: int = 143):
    """Deterministic stratified folds, so both orderings see the same splits."""
    return list(StratifiedKFold(n_splits=n_splits, shuffle=True, random_state=seed).split(X, y))


def select_then_split_score(X, y, k: int = 20, n_splits: int = 5, seed: int = 143) -> float:
    """The WRONG order: choose features using every row, then cross-validate.

    Every fold's test rows helped choose the features, so the features were
    fitted to the answers. Nothing raises. A number comes out.
    """
    chosen = top_k_features(X, y, k)
    scores = []
    for train, test in folds(X, y, n_splits, seed):
        model = KNeighborsClassifier(1).fit(X[train][:, chosen], y[train])
        scores.append(float(np.mean(model.predict(X[test][:, chosen]) == y[test])))
    return float(np.mean(scores))


def split_then_select_score(X, y, k: int = 20, n_splits: int = 5, seed: int = 143) -> float:
    """The RIGHT order: split first, then choose features inside each fold."""
    scores = []
    for train, test in folds(X, y, n_splits, seed):
        chosen = top_k_features(X[train], y[train], k)
        model = KNeighborsClassifier(1).fit(X[train][:, chosen], y[train])
        scores.append(float(np.mean(model.predict(X[test][:, chosen]) == y[test])))
    return float(np.mean(scores))


def inflation_by_k(X, y, ks, n_splits: int = 5, seed: int = 143):
    """How much the wrong order inflates the score, at each feature count."""
    rows = []
    for k in ks:
        wrong = select_then_split_score(X, y, k, n_splits, seed)
        right = split_then_select_score(X, y, k, n_splits, seed)
        rows.append((k, round(wrong, 4), round(right, 4), round(wrong - right, 4)))
    return rows


# --------------------------------------------------------------------------
# Metrics, baselines and error analysis
# --------------------------------------------------------------------------


def candidate_models() -> dict:
    """The five candidates exercise 4 compares. Settings are fixed."""
    return {
        "majority baseline": DummyClassifier(strategy="most_frequent"),
        "logistic (default threshold)": LogisticRegression(max_iter=1000),
        "logistic (balanced)": LogisticRegression(max_iter=1000, class_weight="balanced"),
        "5-NN": KNeighborsClassifier(5),
        "depth-3 tree": DecisionTreeClassifier(max_depth=3, random_state=0),
    }


def score_all(models: dict, X_train, y_train, X_test, y_test) -> dict:
    """Accuracy, precision, recall and F1 for every candidate."""
    out = {}
    for name, model in models.items():
        model.fit(X_train, y_train)
        pred = model.predict(X_test)
        out[name] = {
            "accuracy": round(float(accuracy_score(y_test, pred)), 4),
            "precision": round(float(precision_score(y_test, pred, zero_division=0)), 4),
            "recall": round(float(recall_score(y_test, pred, zero_division=0)), 4),
            "f1": round(float(f1_score(y_test, pred, zero_division=0)), 4),
        }
    return out


def winner(scores: dict, metric: str) -> str:
    """Which candidate a given metric would have you ship."""
    return max(scores, key=lambda name: scores[name][metric])


def error_table(model, X_test, y_test):
    """The confusion matrix, as plain integers -- rows true, columns predicted."""
    return confusion_matrix(y_test, model.predict(X_test)).tolist()


# --------------------------------------------------------------------------
# The stages themselves
# --------------------------------------------------------------------------


def _stage_load(artifact: Artifact) -> dict:
    X, y = noise_dataset(
        artifact.data["n_samples"], artifact.data["n_features"], artifact.data["seed"]
    )
    return {"X": X, "y": y}


def _stage_split(artifact: Artifact) -> dict:
    return {"folds": folds(artifact.data["X"], artifact.data["y"], seed=artifact.data["seed"])}


def _stage_select(artifact: Artifact) -> dict:
    """Choose features. Requires `folds`, so it cannot run before the split."""
    per_fold = []
    for train, _test in artifact.data["folds"]:
        per_fold.append(
            top_k_features(
                artifact.data["X"][train], artifact.data["y"][train], artifact.data["k"]
            )
        )
    return {"selected": per_fold}


def _stage_leaky_select(artifact: Artifact) -> dict:
    """Choose features from every row at once, then reuse them in every fold.

    This is what a leaky pipeline actually does. It is not a strawman: the
    global selection is computed once, which is cheaper, and then handed to
    every fold as though it had been computed there.
    """
    chosen = top_k_features(artifact.data["X"], artifact.data["y"], artifact.data["k"])
    n_folds = len(artifact.data.get("folds", [])) or artifact.data["n_splits"]
    return {"selected": [chosen for _ in range(n_folds)]}


def _stage_fit_and_score(artifact: Artifact) -> dict:
    scores = []
    for (train, test), chosen in zip(artifact.data["folds"], artifact.data["selected"]):
        model = KNeighborsClassifier(1).fit(
            artifact.data["X"][train][:, chosen], artifact.data["y"][train]
        )
        pred = model.predict(artifact.data["X"][test][:, chosen])
        scores.append(float(np.mean(pred == artifact.data["y"][test])))
    return {"fold_scores": np.array(scores), "score": float(np.mean(scores))}


def _stage_baseline(artifact: Artifact) -> dict:
    y = artifact.data["y"]
    counts = np.bincount(y)
    return {"baseline": float(counts.max() / len(y))}


def honest_stages():
    """Load, split, select inside the split, fit, baseline. In that order."""
    return [
        Stage("load", _stage_load, requires=("n_samples", "n_features", "seed"), produces=("X", "y")),
        Stage("split", _stage_split, requires=("X", "y", "seed"), produces=("folds",)),
        Stage("select", _stage_select, requires=("X", "y", "folds", "k"), produces=("selected",)),
        Stage(
            "fit_and_score",
            _stage_fit_and_score,
            requires=("X", "y", "folds", "selected"),
            produces=("fold_scores", "score"),
        ),
        Stage("baseline", _stage_baseline, requires=("y",), produces=("baseline",)),
    ]


def leaky_stages():
    """The same five stages, with selection moved in front of the split.

    Note the `select` stage still declares `folds` among its requirements,
    because that requirement is *true*: choosing features is a per-fold
    operation. Declaring it honestly is the entire mechanism by which the
    runner can notice the ordering is wrong. A team that writes
    `requires=("X", "y")` here has not been caught out by a subtle bug --
    they have written down a claim that is false.
    """
    return [
        Stage("load", _stage_load, requires=("n_samples", "n_features", "seed"), produces=("X", "y")),
        Stage(
            "select",
            _stage_leaky_select,
            requires=("X", "y", "folds", "k"),
            produces=("selected",),
        ),
        Stage("split", _stage_split, requires=("X", "y", "seed"), produces=("folds",)),
        Stage(
            "fit_and_score",
            _stage_fit_and_score,
            requires=("X", "y", "folds", "selected"),
            produces=("fold_scores", "score"),
        ),
        Stage("baseline", _stage_baseline, requires=("y",), produces=("baseline",)),
    ]


def starting_artifact(
    n_samples: int = 100,
    n_features: int = 5000,
    k: int = 20,
    seed: int = 143,
    n_splits: int = 5,
):
    """The inputs the pipeline is given before any stage has run."""
    return Artifact(
        data={
            "n_samples": n_samples,
            "n_features": n_features,
            "k": k,
            "seed": seed,
            "n_splits": n_splits,
        }
    )

lesson_id: D143
day: 143
kind: guided-build
languages:
  - python
  - bash
setup_commands:
  - cd labs/sections/machine-learning/day-143-the-machine-learning-workflow
  - python3 -m venv .venv
  - .venv/bin/pip install -r requirements/requirements.txt
  - >-
    .venv/bin/python3 -c "import numpy, sklearn; print(numpy.__version__,
    sklearn.__version__)"
run_commands:
  - .venv/bin/pytest examples -q
  - .venv/bin/pytest starter -q
  - .venv/bin/python3 examples/report_measurements.py
test_commands:
  - bash tests/run_tests.sh
cleanup_commands:
  - >-
    find . -path ./.venv -prune -o -type d -name '__pycache__' -print -exec rm -rf -- {}
    +
  - rm -rf .pytest_cache
  - 'rm -rf .venv  # optional: removes the lab virtual environment'
  - 'git checkout -- starter/  # optional: reset your work'
requires_network: true
requires_api_key: false
estimated_minutes: 60
last_executed: '2026-08-27'
executed_on: >-
  macOS 26.5.2 (Apple Silicon, arm64), Python 3.14.0, numpy 2.5.2, scikit-learn 1.9.0,
  pytest 9.1.1, bash 3.2.57 -- bash tests/run_tests.sh -> 14 checks, 0 failure(s), exit
  0. pytest examples -q -> 17 passed. pytest starter -q -> 4 passed, 13 skipped (the
  four machinery checks in test_workflow_lib.py are solved in both directories; the
  thirteen exercise stubs in starter/test_workflow_claims.py are untouched). Everything
  ran through a real lab-local .venv created by the documented setup commands;
  scikit-learn pulled in scipy 1.18.1, joblib 1.5.3 and threadpoolctl 3.6.0 as its own
  dependencies, none of which this lab imports directly. The lab is fully offline after
  the pip install -- every dataset is generated on the spot from a seeded
  numpy.random.default_rng, nothing is downloaded, no dataset is bundled, and harness
  check 9 confirms no URL appears anywhere in starter/ or examples/ source. Section 7 of
  the harness copies examples/ into a mktemp-d scratch directory, confirms 17 passed,
  rewrites `assert leaky.data["score"] == 0.73` to 0.50, confirms a non-zero exit naming
  the failing test, and removes the scratch directory. Separately, by hand, `assert
  result.data["baseline"] == 0.54` was changed to 0.99 in
  examples/test_workflow_claims.py and the whole harness re-run: it reported 14 checks,
  2 failure(s) and exited 1 (both the pytest run and the pytest-free direct reproduction
  in section 2 caught it); the file was restored and the harness returned to 14 checks,
  0 failure(s), exit 0. MEASURED PAIRS, all captured verbatim in
  expected-output/measured-values.txt. (1) The pipeline runs as five declared stages
  leaving a step log -- load produces X and y, split produces folds, select produces
  selected, fit_and_score produces fold_scores and score, baseline produces baseline --
  and no stage mutates the artifact it was given, verified by checking the starting
  artifact is untouched after a full run. (2) On 100 rows of 5000 pure-noise features
  whose labels are coin flips, the honest pipeline scores exactly 0.5000 against a
  majority baseline of 0.5400, with per-fold scores [0.5, 0.55, 0.5, 0.4, 0.55]. (3) THE
  CENTRAL RESULT: transposing exactly two stages -- moving select in front of split --
  raises that score to 0.7300 with contracts disabled, inventing 0.2300 of accuracy on
  data that contains nothing to learn, with no warning of any kind; the step logs differ
  by one transposition and nothing else. With contracts enforced the same pipeline
  raises StageContractError naming stage 'select' and the missing key 'folds', and
  harness check 8 confirms it does so at five different seeds rather than once. (4) The
  inflation grows with the number of features chosen: at k = 5, 10, 20 and 50 the wrong
  order scores 0.65, 0.72, 0.73 and 0.85 against honest scores of 0.39, 0.50, 0.50 and
  0.38, inventing 0.26, 0.22, 0.23 and 0.47. (5) On an 8-percent-positive imbalanced
  problem the metric decides which model ships: a majority-class baseline scores 0.9200
  accuracy with 0.0000 recall, logistic regression at its default threshold scores
  0.9435 accuracy and 0.4813 recall, the same model with balanced class weights scores
  0.8685 accuracy and 0.8438 recall, 5-NN scores 0.9360 and a depth-3 tree 0.9275 --
  accuracy and precision pick the default-threshold logistic while recall picks the
  balanced one, so the decision inverts with nothing changing but the metric. Three of
  the four real models beat the constant, by at most 0.0235, and the only model that
  finds most of the positives scores worse than the constant. (6) Error analysis on that
  94.35-percent-accurate model gives the confusion matrix [[1810, 30], [83, 77]]: it
  misses 83 positives while catching 77, so it misses more of the thing you care about
  than it finds. (7) Two runs of the honest pipeline at seed 143 produce byte-identical
  manifests of SHA-256 content hashes while a run at seed 144 does not, which is the
  control that makes the first claim mean something. (8) The fitting stage is 9 of the
  pipeline's 30 lines, 30.00 percent -- and this pipeline has no cleaning, monitoring or
  deployment stage at all, so that figure is reported as an upper bound rather than an
  estimate. THREE HONESTY CALLS. FIRST: two of the honest scores in the inflation table
  are BELOW chance, at 0.39 and 0.38. That is not anti-learning; 20 test rows per fold
  gives a standard error near 0.11. The lab asserts right <= 0.5 rather than right ==
  0.5, because the structural claim is the defensible one, and FIELDS.md says so
  explicitly. SECOND: the stage line counts are measured from this lab's own source with
  inspect.getsource and will legitimately change if the library is edited; the lesson
  reports 30 percent as an upper bound on a pipeline with no cleaning, monitoring or
  deployment stage rather than repeating the familiar folklore percentage, which could
  not be verified here. THIRD: the four manifest hashes are content hashes over exact
  float bytes and therefore move with the NumPy pin; the property asserted as
  version-independent is that two runs at one seed agree and two seeds do not, with the
  literal hashes kept only as a drift detector on this machine.

# Requirements

`requirements.txt` pins the three packages this lab imports directly, at
the exact versions the captured output in `expected-output/` was produced
with:

```
numpy==2.5.2
scikit-learn==1.9.0
pytest==9.1.1
```

Installing scikit-learn also pulls in scipy, joblib and threadpoolctl as
its own dependencies. This lab imports none of them directly and does not
pin them; the versions present during capture are recorded in
`../expected-output/FIELDS.md`.

## Why the versions are pinned exactly

Most numbers in this lab come from a seeded `numpy.random.default_rng`,
and NumPy's documentation is explicit that `Generator` makes no promise of
stream compatibility between versions. A different NumPy can legitimately
produce a different stream from the same seed, and every sampled figure
would move — including the four manifest hashes in exercise 6, which are
content hashes over those exact float bytes.

The structural results do not depend on the pins at all: the step logs,
the `StageContractError`, the direction of the inflation, and the fact
that two runs at one seed agree. `expected-output/FIELDS.md` separates the
two categories explicitly, and it is worth reading before you conclude
that a mismatch is a bug.

## Installing

From the lab directory:

```bash
python3 -m venv .venv
.venv/bin/pip install -r requirements/requirements.txt
```

The install step needs the network. Everything after it is offline: every
dataset in this lab is generated on the spot from a seeded generator, and
nothing is downloaded.

## Free and open-source status

All three packages are free and open source — NumPy and scikit-learn under
the BSD 3-Clause licence, pytest under the MIT licence. There is no paid
tier, no account and no API key anywhere in this lab.

numpy==2.5.2
scikit-learn==1.9.0
pytest==9.1.1

# Day 143 lab brief — The Workflow, Wired Up

The machine learning workflow is normally drawn as a row of boxes with
arrows between them. Everybody nods. Then everybody goes and writes a
notebook where the boxes are cells, the arrows are the order you happened
to run them in, and the whole thing produces a number that nobody can
reproduce and nobody can defend.

This lab builds the same workflow with the arrows made load-bearing.
Every stage declares **what it requires** and **what it produces**, the
runner refuses to run a stage whose inputs are absent, and every run
leaves a step log and a manifest of content hashes behind it.

## The claim you are here to measure

> The same stages in a different order give a different answer, and the
> wrong order is silent.

Exercise 3 does it in five lines. A dataset of 100 rows and 5000
features, where the labels are coin flips and no feature carries any
information whatsoever:

| Pipeline | Order | Score |
| --- | --- | --- |
| honest | load, **split, select**, fit, baseline | **0.50** |
| leaky, contracts off | load, **select, split**, fit, baseline | **0.73** |
| leaky, contracts on | same as above | `StageContractError` |

Twenty-three accuracy points, on data that contains nothing to learn,
produced by transposing two stages. Same data, same model, same folds,
same seed. Nothing raises. Nothing warns. A number comes out and it looks
like a result.

And then the third row, which is the actual lesson:

```
StageContractError: stage 'select' requires ['folds'] which no earlier
stage produced
```

The contract is not ceremony. It is what turns a silent twenty-three
point lie into a loud error naming the stage that broke.

## The thing to understand about that contract

Look carefully at how `leaky_stages()` is written. Its `select` stage
still declares `folds` among its requirements — because that requirement
is **true**. Choosing features is a per-fold operation. Declaring it
honestly is the entire mechanism by which the runner can notice.

A team that writes `requires=("X", "y")` on that stage has not been caught
out by a subtle bug. They have written down a claim that is false, and
every checking tool in the world is downstream of that.

## Six more things this lab measures

| # | What it establishes |
| --- | --- |
| 1 | The pipeline as stages with a step log, and stages that never mutate their input |
| 2 | The honest pipeline reports chance on data that is chance |
| 3b | The inflation grows with the number of features chosen: +0.26, +0.22, +0.23, +0.47 |
| 4 | The metric decides which model you ship — accuracy and recall pick different winners |
| 5 | A 94.35 percent accurate model that misses more positives than it catches |
| 6 | Two runs of the pipeline are byte-identical; a different seed is not |
| 7 | The modelling stage is 30 percent of this pipeline, and that is an upper bound |

Exercise 4 is the one to sit with. On an imbalanced problem — 8 percent
positive — a majority-class baseline scores **0.92 accuracy with zero
recall**, and three of the four real models beat it by at most 2.35
points. The one model that actually finds most of the positives, at 0.8438
recall, scores **worse than the constant** on accuracy.

Choosing the metric is therefore not a reporting decision made at the end.
It is the decision that determines which model you ship, and you make it
before any model exists.

## How to work

1. Build the environment (see the lab `README.md`).
2. Run `.venv/bin/pytest starter -q`. You will see four passes (the
   machinery checks in `test_workflow_lib.py`) and thirteen skips.
3. Replace one `pytest.skip(...)` at a time with real code. The skip text
   names the exact helper and the exact value to assert.
4. Print the measured pair in every exercise. A number you did not print
   is a number you did not look at.
5. When you want the whole measured table at once, run
   `.venv/bin/python3 examples/report_measurements.py`.

Do not run `pytest starter examples` in one invocation. Both directories
define `workflow_lib.py`, `test_workflow_lib.py` and
`test_workflow_claims.py`; pytest aborts on the module-name collision.
Run them separately, always.

## A note on the honest score being 0.50

It is 0.50 because the labels are coin flips, and that is the correct
answer. But the five per-fold scores behind it are `[0.5, 0.55, 0.5, 0.4,
0.55]`, which is a wide spread — twenty test rows per fold buys you a
standard error of roughly 0.11.

Two of the honest scores in the `inflation_by_k` table are *below* chance,
at 0.39 and 0.38. That is not anti-learning and it is not a bug. An
estimate of a 0.5 quantity from a small sample wanders, and it wanders
below as readily as above. The lab therefore asserts `right <= 0.5`
rather than `right == 0.5`, because the structural claim is the one worth
asserting.

"""Thirteen exercises in what the machine learning workflow actually is.

Read `00_brief.md` first. Each function below is a `pytest.skip` naming
exactly what to build and what to assert; replace the skip with real code.
`workflow_lib.py` is complete -- it is the machinery, not the exercise.

Run this suite on its own:

    .venv/bin/pytest starter -q

Never run `pytest starter examples` in one invocation: both directories
define modules with the same names and pytest aborts on the collision.
"""

import numpy as np  # noqa: F401  (you will need it)
import pytest

import workflow_lib as w  # noqa: F401  (you will need it)


@pytest.fixture(scope="module")
def noise():
    return w.noise_dataset()


@pytest.fixture(scope="module")
def imbalanced():
    return w.imbalanced_dataset(1000, 11), w.imbalanced_dataset(2000, 12)


def test_01_the_honest_pipeline_runs_in_the_declared_order():
    pytest.skip(
        "Run w.run_pipeline(w.honest_stages(), w.starting_artifact()). Assert "
        "the stage names in result.log are load, split, select, fit_and_score, "
        "baseline in that order, and that each stage recorded exactly the keys "
        "it declared: load -> ('X', 'y'), split -> ('folds',), select -> "
        "('selected',), fit_and_score -> ('fold_scores', 'score'), baseline -> "
        "('baseline',)."
    )


def test_01b_a_stage_never_mutates_the_artifact_it_was_given():
    pytest.skip(
        "Capture set(start.data) before running the pipeline, run it, and "
        "assert the starting artifact's keys and empty log are unchanged. A "
        "stage that mutates its input makes the step log a work of fiction, "
        "because the log then describes states that no longer exist."
    )


def test_02_the_honest_pipeline_reports_chance_on_pure_noise():
    pytest.skip(
        "Assert the honest pipeline's score is exactly 0.5, its baseline is "
        "0.54, and fold_scores has shape (5,). The labels in noise_dataset "
        "are coin flips, so chance is the only honest answer -- and note the "
        "baseline of 0.54 is above 0.5 because 100 coin flips do not land "
        "exactly fifty-fifty."
    )


def test_03_reordering_two_stages_invents_twenty_three_accuracy_points():
    pytest.skip(
        "Run the honest pipeline, then run w.leaky_stages() with "
        "enforce_contracts=False. Assert the scores are 0.5 and 0.73 and that "
        "the difference is 0.23. Then assert the two step logs differ by "
        "exactly one transposition: select and split have swapped places and "
        "nothing else changed. Same data, same model, same folds."
    )


def test_03b_the_contract_turns_a_silent_lie_into_a_named_failure():
    pytest.skip(
        "Run w.leaky_stages() with enforce_contracts=True inside "
        "pytest.raises(w.StageContractError). Assert the message names the "
        "stage 'select' and the missing key 'folds'. Then assert the honest "
        "pipeline passes the very same contracts and still scores 0.5."
    )


def test_03c_the_inflation_grows_with_the_number_of_features_chosen(noise):
    pytest.skip(
        "Assert w.inflation_by_k(X, y, [5, 10, 20, 50]) equals [(5, 0.65, "
        "0.39, 0.26), (10, 0.72, 0.5, 0.22), (20, 0.73, 0.5, 0.23), (50, "
        "0.85, 0.38, 0.47)]. Then assert the structural facts that hold "
        "regardless of the numbers: every wrong-order score beats its honest "
        "counterpart, and every honest score is at or below chance."
    )


def test_04_the_metric_you_choose_decides_which_model_you_ship(imbalanced):
    pytest.skip(
        "Score w.candidate_models() with w.score_all. Assert the majority "
        "baseline is accuracy 0.92 with recall 0.0, that logistic at the "
        "default threshold is 0.9435 accuracy and 0.4813 recall, and that "
        "logistic balanced is 0.8685 accuracy and 0.8438 recall. Then assert "
        "w.winner picks 'logistic (default threshold)' on accuracy and "
        "'logistic (balanced)' on recall -- the decision inverts, with "
        "nothing changing but the metric."
    )


def test_04b_a_model_that_never_predicts_the_positive_class_scores_ninety_two(imbalanced):
    pytest.skip(
        "Assert the majority baseline scores 0.92 accuracy and 0.0 recall. "
        "Collect the models that beat 0.92 accuracy and assert the set is "
        "exactly logistic (default threshold), 5-NN and depth-3 tree, and "
        "that the best of them beats the constant by only 0.0235. Then "
        "assert the one model that actually finds positives -- logistic "
        "balanced, recall 0.8438 -- scores WORSE than the constant. Assert "
        "the test set carries 160 positives in 2000 rows."
    )


def test_05_the_confusion_matrix_says_what_the_accuracy_hides(imbalanced):
    pytest.skip(
        "Fit logistic at the default threshold and assert w.error_table is "
        "[[1810, 30], [83, 77]]. Assert the four cells sum to 2000, that the "
        "false negatives (83) exceed the true positives (77), and that the "
        "accuracy recomputed from the table is 0.9435. A model can be 94 "
        "percent accurate and miss more of the thing you care about than it "
        "finds."
    )


def test_06_two_runs_of_the_pipeline_are_byte_identical():
    pytest.skip(
        "Build w.manifest over ('X', 'y', 'fold_scores', 'score') for two "
        "separate runs at seed 143 and assert they are equal. Assert the "
        "manifest is {'X': '51b0a421bd652dd2', 'fold_scores': "
        "'8f0ac332958b9bc4', 'score': 'd2cbad71ff333de6', 'y': "
        "'9984503b5352c5a1'}. A pipeline that cannot prove it produced the "
        "same thing twice cannot be debugged."
    )


def test_06b_a_different_seed_produces_a_different_manifest():
    pytest.skip(
        "Compare the seed 143 manifest against a seed 144 one and assert "
        "they differ, including on the 'X' key specifically. This is the "
        "control: a manifest that never changes is not evidence of "
        "determinism, it is evidence that you are hashing a constant."
    )


def test_07_the_modelling_stage_is_the_smallest_part_of_the_pipeline():
    pytest.skip(
        "Assert w.stage_source_lines(w.honest_stages()) is {'load': 5, "
        "'split': 2, 'select': 10, 'fit_and_score': 9, 'baseline': 4}, "
        "totalling 30, and that fit_and_score is 30 percent of it. Then "
        "assert what is NOT in that dict: no cleaning stage, no monitoring "
        "stage, no deployment stage. That is why 30 percent is an upper "
        "bound and not an estimate."
    )


def test_08_a_stage_cannot_run_without_the_inputs_it_declared():
    pytest.skip(
        "Run the honest pipeline against an empty w.Artifact(data={}) inside "
        "pytest.raises(w.StageContractError) and assert the message names "
        "'load'. Then run the same thing with enforce_contracts=False and "
        "assert it raises KeyError instead. Both fail; only one tells you "
        "which stage broke."
    )

"""Machinery checks: the runner itself behaves, before any claim is made.

These four tests are solved in both `starter/` and `examples/`. They exist
so that a broken runner reports itself as a broken runner rather than as a
surprising scientific result.
"""

import numpy as np
import pytest

import workflow_lib as w


def test_the_artifact_is_immutable_and_carries_its_history():
    start = w.Artifact(data={"a": 1})
    nxt = start.with_(b=2)
    assert start.data == {"a": 1}
    assert nxt.data == {"a": 1, "b": 2}
    assert nxt is not start
    # Overwriting a key is allowed; silently mutating the original is not.
    third = nxt.with_(a=99)
    assert third.data["a"] == 99 and nxt.data["a"] == 1


def test_the_fingerprint_is_stable_content_addressing():
    a = np.arange(10)
    b = np.arange(10)
    assert w.fingerprint(a) == w.fingerprint(b)
    assert w.fingerprint(a) != w.fingerprint(np.arange(11))
    # dtype is part of the identity: the same values in a different type
    # are a different artifact, and treating them as equal hides real bugs.
    assert w.fingerprint(np.arange(10)) != w.fingerprint(np.arange(10, dtype=float))
    assert len(w.fingerprint(a)) == 16


def test_the_contract_checks_both_directions():
    called = []

    def under_producing(_artifact):
        called.append("under")
        return {}

    stages = [w.Stage("under", under_producing, requires=(), produces=("x",))]
    with pytest.raises(w.StageContractError) as excinfo:
        w.run_pipeline(stages, w.Artifact(data={}))
    assert "'under'" in str(excinfo.value)
    assert called == ["under"]

    def over_producing(_artifact):
        return {"x": 1, "surprise": 2}

    stages = [w.Stage("over", over_producing, requires=(), produces=("x",))]
    with pytest.raises(w.StageContractError):
        w.run_pipeline(stages, w.Artifact(data={}))


def test_the_folds_are_stratified_and_cover_every_row_exactly_once():
    X, y = w.imbalanced_dataset(500, seed=3)
    splits = w.folds(X, y, n_splits=5, seed=143)
    assert len(splits) == 5
    seen = np.concatenate([test for _train, test in splits])
    assert sorted(seen.tolist()) == list(range(500))
    # Stratified: every fold carries roughly the population positive rate.
    rate = float(y.mean())
    for _train, test in splits:
        assert abs(float(y[test].mean()) - rate) < 0.02

"""The machine learning workflow as runnable stages, with contracts.

The workflow in the textbook is a row of boxes with arrows. This module is
the same workflow with the arrows made load-bearing: every stage declares
what it requires and what it produces, the runner refuses to run a stage
whose inputs are absent, and every run leaves a step log and a manifest of
content hashes behind it.

That is not ceremony. The point of this lab is that **the same three
stages in a different order produce a different number**, and that the
wrong order is silent. A stage contract is what turns a silent
twenty-three point lie into a loud error naming the stage that broke.
"""

from __future__ import annotations

import hashlib
import inspect
from dataclasses import dataclass, field
from typing import Callable

import numpy as np

from sklearn.dummy import DummyClassifier
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import (
    accuracy_score,
    confusion_matrix,
    f1_score,
    precision_score,
    recall_score,
)
from sklearn.model_selection import StratifiedKFold
from sklearn.neighbors import KNeighborsClassifier
from sklearn.tree import DecisionTreeClassifier


class StageContractError(RuntimeError):
    """Raised when a stage is asked to run without the inputs it declared."""


# --------------------------------------------------------------------------
# The runner
# --------------------------------------------------------------------------


@dataclass
class Artifact:
    """Everything the pipeline knows so far, plus how it came to know it."""

    data: dict = field(default_factory=dict)
    log: list = field(default_factory=list)

    def with_(self, **produced) -> "Artifact":
        """Return a new artifact carrying the additional keys.

        Deliberately not in-place. A stage that mutates its input makes the
        step log a work of fiction, because the log then describes states
        that no longer exist.
        """
        merged = dict(self.data)
        merged.update(produced)
        return Artifact(data=merged, log=list(self.log))


@dataclass
class Stage:
    """One step of the workflow, with its input and output contract."""

    name: str
    run: Callable[[Artifact], dict]
    requires: tuple = ()
    produces: tuple = ()


def run_pipeline(stages, artifact: Artifact, enforce_contracts: bool = True) -> Artifact:
    """Run every stage in order, recording what each one did.

    With ``enforce_contracts=False`` the runner behaves like most real
    pipelines: it runs whatever it is given, in whatever order, and reports
    a number. Exercise 3 measures what that costs.
    """
    for stage in stages:
        if enforce_contracts:
            missing = [k for k in stage.requires if k not in artifact.data]
            if missing:
                raise StageContractError(
                    f"stage {stage.name!r} requires {missing} which no earlier stage produced"
                )
        produced = stage.run(artifact)
        if enforce_contracts:
            unexpected = sorted(set(produced) - set(stage.produces))
            absent = sorted(set(stage.produces) - set(produced))
            if absent or unexpected:
                raise StageContractError(
                    f"stage {stage.name!r} declared {list(stage.produces)} "
                    f"but produced {sorted(produced)}"
                )
        artifact = artifact.with_(**produced)
        artifact.log.append((stage.name, tuple(sorted(produced))))
    return artifact


def step_log(artifact: Artifact):
    """The ordered record of which stage produced which keys."""
    return list(artifact.log)


def fingerprint(value) -> str:
    """A stable content hash for an array, a number or a string."""
    if isinstance(value, np.ndarray):
        payload = np.ascontiguousarray(value).tobytes() + str(value.dtype).encode()
    else:
        payload = repr(value).encode()
    return hashlib.sha256(payload).hexdigest()[:16]


def manifest(artifact: Artifact, keys) -> dict:
    """Content hashes for the named keys, so two runs can be compared.

    A pipeline that cannot prove it produced the same thing twice cannot be
    debugged, because you can never tell a fix from a coincidence.
    """
    return {k: fingerprint(artifact.data[k]) for k in sorted(keys)}


def stage_source_lines(stages) -> dict:
    """How many lines of code each stage's function actually is.

    Self-referential on purpose: exercise 7 uses this to measure the shape
    of this very pipeline, rather than repeating the folklore figure about
    how much of the work is not modelling.
    """
    out = {}
    for stage in stages:
        source = inspect.getsource(stage.run)
        lines = [ln for ln in source.splitlines() if ln.strip() and not ln.strip().startswith("#")]
        out[stage.name] = len(lines)
    return out


# --------------------------------------------------------------------------
# Datasets
# --------------------------------------------------------------------------


def noise_dataset(n_samples: int = 100, n_features: int = 5000, seed: int = 143):
    """Labels are coin flips. No feature carries any information whatsoever.

    Any workflow that reports better than chance on this data has a bug,
    and the bug is what exercise 3 is about.
    """
    rng = np.random.default_rng(seed)
    X = rng.normal(size=(n_samples, n_features))
    y = rng.integers(0, 2, size=n_samples)
    return X, y


def imbalanced_dataset(n: int, seed: int, rate: float = 0.08):
    """A rare positive class whose features overlap the negatives.

    Eight percent positive, shifted by 1.1 standard deviations. Separable
    enough to be worth modelling, overlapping enough that the metric you
    choose decides which model wins.
    """
    rng = np.random.default_rng(seed)
    n_pos = int(round(n * rate))
    y = np.zeros(n, dtype=int)
    y[:n_pos] = 1
    X = rng.normal(size=(n, 4))
    X[y == 1] += 1.1
    order = rng.permutation(n)
    return X[order], y[order]


# --------------------------------------------------------------------------
# The individual steps, usable on their own
# --------------------------------------------------------------------------


def correlation_ranking(X, y):
    """Absolute correlation between each column and the label."""
    y = np.asarray(y, dtype=float)
    out = np.empty(X.shape[1])
    for j in range(X.shape[1]):
        out[j] = abs(float(np.corrcoef(X[:, j], y)[0, 1]))
    return out


def top_k_features(X, y, k: int):
    """The k columns most correlated with the label, as indices."""
    return np.argsort(correlation_ranking(X, y))[-k:]


def folds(X, y, n_splits: int = 5, seed: int = 143):
    """Deterministic stratified folds, so both orderings see the same splits."""
    return list(StratifiedKFold(n_splits=n_splits, shuffle=True, random_state=seed).split(X, y))


def select_then_split_score(X, y, k: int = 20, n_splits: int = 5, seed: int = 143) -> float:
    """The WRONG order: choose features using every row, then cross-validate.

    Every fold's test rows helped choose the features, so the features were
    fitted to the answers. Nothing raises. A number comes out.
    """
    chosen = top_k_features(X, y, k)
    scores = []
    for train, test in folds(X, y, n_splits, seed):
        model = KNeighborsClassifier(1).fit(X[train][:, chosen], y[train])
        scores.append(float(np.mean(model.predict(X[test][:, chosen]) == y[test])))
    return float(np.mean(scores))


def split_then_select_score(X, y, k: int = 20, n_splits: int = 5, seed: int = 143) -> float:
    """The RIGHT order: split first, then choose features inside each fold."""
    scores = []
    for train, test in folds(X, y, n_splits, seed):
        chosen = top_k_features(X[train], y[train], k)
        model = KNeighborsClassifier(1).fit(X[train][:, chosen], y[train])
        scores.append(float(np.mean(model.predict(X[test][:, chosen]) == y[test])))
    return float(np.mean(scores))


def inflation_by_k(X, y, ks, n_splits: int = 5, seed: int = 143):
    """How much the wrong order inflates the score, at each feature count."""
    rows = []
    for k in ks:
        wrong = select_then_split_score(X, y, k, n_splits, seed)
        right = split_then_select_score(X, y, k, n_splits, seed)
        rows.append((k, round(wrong, 4), round(right, 4), round(wrong - right, 4)))
    return rows


# --------------------------------------------------------------------------
# Metrics, baselines and error analysis
# --------------------------------------------------------------------------


def candidate_models() -> dict:
    """The five candidates exercise 4 compares. Settings are fixed."""
    return {
        "majority baseline": DummyClassifier(strategy="most_frequent"),
        "logistic (default threshold)": LogisticRegression(max_iter=1000),
        "logistic (balanced)": LogisticRegression(max_iter=1000, class_weight="balanced"),
        "5-NN": KNeighborsClassifier(5),
        "depth-3 tree": DecisionTreeClassifier(max_depth=3, random_state=0),
    }


def score_all(models: dict, X_train, y_train, X_test, y_test) -> dict:
    """Accuracy, precision, recall and F1 for every candidate."""
    out = {}
    for name, model in models.items():
        model.fit(X_train, y_train)
        pred = model.predict(X_test)
        out[name] = {
            "accuracy": round(float(accuracy_score(y_test, pred)), 4),
            "precision": round(float(precision_score(y_test, pred, zero_division=0)), 4),
            "recall": round(float(recall_score(y_test, pred, zero_division=0)), 4),
            "f1": round(float(f1_score(y_test, pred, zero_division=0)), 4),
        }
    return out


def winner(scores: dict, metric: str) -> str:
    """Which candidate a given metric would have you ship."""
    return max(scores, key=lambda name: scores[name][metric])


def error_table(model, X_test, y_test):
    """The confusion matrix, as plain integers -- rows true, columns predicted."""
    return confusion_matrix(y_test, model.predict(X_test)).tolist()


# --------------------------------------------------------------------------
# The stages themselves
# --------------------------------------------------------------------------


def _stage_load(artifact: Artifact) -> dict:
    X, y = noise_dataset(
        artifact.data["n_samples"], artifact.data["n_features"], artifact.data["seed"]
    )
    return {"X": X, "y": y}


def _stage_split(artifact: Artifact) -> dict:
    return {"folds": folds(artifact.data["X"], artifact.data["y"], seed=artifact.data["seed"])}


def _stage_select(artifact: Artifact) -> dict:
    """Choose features. Requires `folds`, so it cannot run before the split."""
    per_fold = []
    for train, _test in artifact.data["folds"]:
        per_fold.append(
            top_k_features(
                artifact.data["X"][train], artifact.data["y"][train], artifact.data["k"]
            )
        )
    return {"selected": per_fold}


def _stage_leaky_select(artifact: Artifact) -> dict:
    """Choose features from every row at once, then reuse them in every fold.

    This is what a leaky pipeline actually does. It is not a strawman: the
    global selection is computed once, which is cheaper, and then handed to
    every fold as though it had been computed there.
    """
    chosen = top_k_features(artifact.data["X"], artifact.data["y"], artifact.data["k"])
    n_folds = len(artifact.data.get("folds", [])) or artifact.data["n_splits"]
    return {"selected": [chosen for _ in range(n_folds)]}


def _stage_fit_and_score(artifact: Artifact) -> dict:
    scores = []
    for (train, test), chosen in zip(artifact.data["folds"], artifact.data["selected"]):
        model = KNeighborsClassifier(1).fit(
            artifact.data["X"][train][:, chosen], artifact.data["y"][train]
        )
        pred = model.predict(artifact.data["X"][test][:, chosen])
        scores.append(float(np.mean(pred == artifact.data["y"][test])))
    return {"fold_scores": np.array(scores), "score": float(np.mean(scores))}


def _stage_baseline(artifact: Artifact) -> dict:
    y = artifact.data["y"]
    counts = np.bincount(y)
    return {"baseline": float(counts.max() / len(y))}


def honest_stages():
    """Load, split, select inside the split, fit, baseline. In that order."""
    return [
        Stage("load", _stage_load, requires=("n_samples", "n_features", "seed"), produces=("X", "y")),
        Stage("split", _stage_split, requires=("X", "y", "seed"), produces=("folds",)),
        Stage("select", _stage_select, requires=("X", "y", "folds", "k"), produces=("selected",)),
        Stage(
            "fit_and_score",
            _stage_fit_and_score,
            requires=("X", "y", "folds", "selected"),
            produces=("fold_scores", "score"),
        ),
        Stage("baseline", _stage_baseline, requires=("y",), produces=("baseline",)),
    ]


def leaky_stages():
    """The same five stages, with selection moved in front of the split.

    Note the `select` stage still declares `folds` among its requirements,
    because that requirement is *true*: choosing features is a per-fold
    operation. Declaring it honestly is the entire mechanism by which the
    runner can notice the ordering is wrong. A team that writes
    `requires=("X", "y")` here has not been caught out by a subtle bug --
    they have written down a claim that is false.
    """
    return [
        Stage("load", _stage_load, requires=("n_samples", "n_features", "seed"), produces=("X", "y")),
        Stage(
            "select",
            _stage_leaky_select,
            requires=("X", "y", "folds", "k"),
            produces=("selected",),
        ),
        Stage("split", _stage_split, requires=("X", "y", "seed"), produces=("folds",)),
        Stage(
            "fit_and_score",
            _stage_fit_and_score,
            requires=("X", "y", "folds", "selected"),
            produces=("fold_scores", "score"),
        ),
        Stage("baseline", _stage_baseline, requires=("y",), produces=("baseline",)),
    ]


def starting_artifact(
    n_samples: int = 100,
    n_features: int = 5000,
    k: int = 20,
    seed: int = 143,
    n_splits: int = 5,
):
    """The inputs the pipeline is given before any stage has run."""
    return Artifact(
        data={
            "n_samples": n_samples,
            "n_features": n_features,
            "k": k,
            "seed": seed,
            "n_splits": n_splits,
        }
    )

#!/usr/bin/env bash
# Day 143 lab harness: "The Workflow, Wired Up"
#
# Prints "N checks, M failure(s)" and exits 0 only when M is zero.
set -u

LAB_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
cd "$LAB_DIR"

PYTHON="${PYTHON:-.venv/bin/python3}"
PYTEST="${PYTEST:-.venv/bin/pytest}"

# Clear caches at the START so the final cleanliness check measures what
# THIS run left behind, not what a previous `pytest starter -q` left.
find . -path ./.venv -prune -o -type d -name '__pycache__' -exec rm -rf -- {} + 2>/dev/null
rm -rf .pytest_cache

CHECKS=0
FAILURES=0

ok() {
  CHECKS=$((CHECKS + 1))
  echo "  ok: $1"
}

fail() {
  CHECKS=$((CHECKS + 1))
  FAILURES=$((FAILURES + 1))
  echo "  FAIL: $1"
}

if [ ! -x "$PYTHON" ]; then
  echo "No lab .venv found at $PYTHON."
  echo "Run: python3 -m venv .venv && .venv/bin/pip install -r requirements/requirements.txt"
  exit 2
fi

echo "1. Installed versions match requirements/requirements.txt"
VERSION_CHECK=$("$PYTHON" - <<'PYEOF'
import numpy, sklearn, pytest
print("numpy", numpy.__version__)
print("scikit-learn", sklearn.__version__)
print("pytest", pytest.__version__)
PYEOF
)
echo "$VERSION_CHECK" | sed 's/^/    /'
while read -r pkg pin; do
  pin_version="${pin#*==}"
  installed=$(echo "$VERSION_CHECK" | awk -v p="$pkg" '$1==p {print $2}')
  if [ "$installed" = "$pin_version" ]; then
    ok "$pkg $installed matches the pin"
  else
    fail "$pkg installed=$installed pinned=$pin_version"
  fi
done < <(sed 's/==/ ==/' requirements/requirements.txt)

echo ""
echo "2. Every published claim, reproduced directly (no pytest involved)"
DIRECT_CHECK=$("$PYTHON" - <<'PYEOF'
import sys
sys.path.insert(0, "examples")

import numpy as np

import workflow_lib as w

errors = []


def expect(label, got, want):
    if got != want:
        errors.append(f"{label}: expected {want}, got {got}")


# 1. The pipeline as stages with a step log
honest = w.run_pipeline(w.honest_stages(), w.starting_artifact())
expect(
    "honest step log",
    [name for name, _ in honest.log],
    ["load", "split", "select", "fit_and_score", "baseline"],
)
expect("load produces", dict(honest.log)["load"], ("X", "y"))
expect("fit produces", dict(honest.log)["fit_and_score"], ("fold_scores", "score"))

# 1b. Stages do not mutate their input
start = w.starting_artifact()
before = set(start.data)
w.run_pipeline(w.honest_stages(), start)
expect("starting artifact untouched", set(start.data), before)
expect("starting artifact log empty", start.log, [])

# 2. Chance on noise
expect("honest score on pure noise", honest.data["score"], 0.5)
expect("majority baseline", honest.data["baseline"], 0.54)
expect("fold scores", honest.data["fold_scores"].tolist(), [0.5, 0.55, 0.5, 0.4, 0.55])

# 3. The transposition, and the contract
leaky = w.run_pipeline(w.leaky_stages(), w.starting_artifact(), enforce_contracts=False)
expect("leaky score, contracts off", leaky.data["score"], 0.73)
expect("accuracy invented by reordering", round(leaky.data["score"] - honest.data["score"], 4), 0.23)
expect(
    "leaky step log",
    [name for name, _ in leaky.log],
    ["load", "select", "split", "fit_and_score", "baseline"],
)
try:
    w.run_pipeline(w.leaky_stages(), w.starting_artifact(), enforce_contracts=True)
except w.StageContractError as exc:
    message = str(exc)
    if "'select'" not in message or "folds" not in message:
        errors.append(f"contract error did not name the stage and key: {message}")
else:
    errors.append("the leaky pipeline passed its contracts, which it must not")

# 3b. Inflation by k
X, y = w.noise_dataset()
expect(
    "inflation by k",
    w.inflation_by_k(X, y, [5, 10, 20, 50]),
    [(5, 0.65, 0.39, 0.26), (10, 0.72, 0.5, 0.22), (20, 0.73, 0.5, 0.23), (50, 0.85, 0.38, 0.47)],
)

# 4. The metric decides
train = w.imbalanced_dataset(1000, 11)
test = w.imbalanced_dataset(2000, 12)
scores = w.score_all(w.candidate_models(), train[0], train[1], test[0], test[1])
expect("baseline accuracy", scores["majority baseline"]["accuracy"], 0.92)
expect("baseline recall", scores["majority baseline"]["recall"], 0.0)
expect("logistic default accuracy", scores["logistic (default threshold)"]["accuracy"], 0.9435)
expect("logistic default recall", scores["logistic (default threshold)"]["recall"], 0.4813)
expect("logistic balanced accuracy", scores["logistic (balanced)"]["accuracy"], 0.8685)
expect("logistic balanced recall", scores["logistic (balanced)"]["recall"], 0.8438)
expect("5-NN accuracy", scores["5-NN"]["accuracy"], 0.936)
expect("depth-3 tree accuracy", scores["depth-3 tree"]["accuracy"], 0.9275)
expect("winner on accuracy", w.winner(scores, "accuracy"), "logistic (default threshold)")
expect("winner on recall", w.winner(scores, "recall"), "logistic (balanced)")
if w.winner(scores, "accuracy") == w.winner(scores, "recall"):
    errors.append("the metric did not invert the decision, contradicting exercise 4")
beats = {n for n, s in scores.items() if s["accuracy"] > 0.92}
expect("models beating the constant", beats, {"logistic (default threshold)", "5-NN", "depth-3 tree"})
expect("test positives", int(test[1].sum()), 160)

# 5. Error analysis
model = w.candidate_models()["logistic (default threshold)"]
model.fit(train[0], train[1])
table = w.error_table(model, test[0], test[1])
expect("confusion matrix", table, [[1810, 30], [83, 77]])
if table[1][0] <= table[1][1]:
    errors.append("the model did not miss more positives than it caught")

# 6. Reproducibility
keys = ("X", "y", "fold_scores", "score")
first = w.manifest(w.run_pipeline(w.honest_stages(), w.starting_artifact()), keys)
second = w.manifest(w.run_pipeline(w.honest_stages(), w.starting_artifact()), keys)
other = w.manifest(w.run_pipeline(w.honest_stages(), w.starting_artifact(seed=144)), keys)
expect(
    "manifest",
    first,
    {
        "X": "51b0a421bd652dd2",
        "fold_scores": "8f0ac332958b9bc4",
        "score": "d2cbad71ff333de6",
        "y": "9984503b5352c5a1",
    },
)
if first != second:
    errors.append("two runs at the same seed produced different manifests")
if first == other:
    errors.append("a different seed produced an identical manifest")

# 7. Stage sizes
lines = w.stage_source_lines(w.honest_stages())
expect(
    "stage line counts",
    lines,
    {"load": 5, "split": 2, "select": 10, "fit_and_score": 9, "baseline": 4},
)
expect("total lines", sum(lines.values()), 30)
expect("fitting share", round(lines["fit_and_score"] / sum(lines.values()), 4), 0.3)

# 8. A missing input
try:
    w.run_pipeline(w.honest_stages(), w.Artifact(data={}))
except w.StageContractError as exc:
    if "'load'" not in str(exc):
        errors.append(f"empty-artifact error did not name the load stage: {exc}")
else:
    errors.append("an empty artifact passed the contracts, which it must not")
try:
    w.run_pipeline(w.honest_stages(), w.Artifact(data={}), enforce_contracts=False)
except KeyError:
    pass
else:
    errors.append("with contracts off, an empty artifact did not raise KeyError")

if errors:
    for e in errors:
        print("ERROR:", e)
    sys.exit(1)
print("all direct checks passed")
PYEOF
)
if echo "$DIRECT_CHECK" | grep -q "all direct checks passed"; then
  ok "exercises 1-8 reproduced directly against workflow_lib, no pytest involved"
else
  fail "direct library checks failed"
  echo "$DIRECT_CHECK" | sed 's/^/    /'
fi

echo ""
echo "3. examples/ passes in full"
EXAMPLES_OUT=$("$PYTEST" examples -q 2>&1)
if echo "$EXAMPLES_OUT" | tail -1 | grep -qE "^17 passed"; then
  ok "pytest examples -q -> 17 passed"
else
  fail "pytest examples -q did not report 17 passed"
  echo "$EXAMPLES_OUT" | tail -20 | sed 's/^/    /'
fi

echo ""
echo "4. starter/ is an untouched skeleton"
STARTER_OUT=$("$PYTEST" starter -q 2>&1)
if echo "$STARTER_OUT" | tail -1 | grep -qE "4 passed, 13 skipped"; then
  ok "pytest starter -q -> 4 passed, 13 skipped (the machinery checks pass; the thirteen exercises are stubs)"
else
  fail "pytest starter -q did not report 4 passed, 13 skipped"
  echo "$STARTER_OUT" | tail -20 | sed 's/^/    /'
fi

echo ""
echo "5. pytest examples starter (one invocation) aborts on the module-name collision"
COMBINED_OUT=$("$PYTEST" examples starter 2>&1)
if echo "$COMBINED_OUT" | grep -q "import file mismatch"; then
  ok "combined invocation reports import file mismatch, as documented -- never run starter and examples together"
else
  fail "combined invocation did not fail with import file mismatch as expected"
fi

echo ""
echo "6. The report reproduces the captured table exactly"
REPORT_OUT=$("$PYTHON" examples/report_measurements.py 2>&1)
if [ "$REPORT_OUT" = "$(cat expected-output/measured-values.txt)" ]; then
  ok "report_measurements.py output is byte-identical to expected-output/measured-values.txt"
else
  fail "report_measurements.py drifted from expected-output/measured-values.txt"
  echo "$REPORT_OUT" | diff - expected-output/measured-values.txt | head -20 | sed 's/^/    /'
fi

echo ""
echo "7. Proof the harness can fail"
SCRATCH=$(mktemp -d "${TMPDIR:-/tmp}/d143-scratch.XXXXXX")
cp examples/*.py "$SCRATCH"/
SCRATCH_OUT=$("$PYTEST" "$SCRATCH" -q 2>&1)
if echo "$SCRATCH_OUT" | tail -1 | grep -qE "^17 passed"; then
  ok "scratch copy of examples/ passes before it is broken"
else
  fail "scratch copy did not pass before being broken: $(echo "$SCRATCH_OUT" | tail -3)"
fi
"$PYTHON" - "$SCRATCH/test_workflow_claims.py" <<'PYEOF'
import sys
path = sys.argv[1]
text = open(path).read()
needle = 'assert leaky.data["score"] == 0.73'
replacement = 'assert leaky.data["score"] == 0.50'
assert needle in text, "could not find the assertion to break"
open(path, "w").write(text.replace(needle, replacement, 1))
PYEOF
BROKEN_OUT=$("$PYTEST" "$SCRATCH" -q 2>&1)
BROKEN_STATUS=$?
if [ "$BROKEN_STATUS" -ne 0 ] && echo "$BROKEN_OUT" | grep -q "test_03_reordering_two_stages_invents_twenty_three_accuracy_points"; then
  ok "breaking exercise 3's assertion produces a non-zero exit and names the failing test"
else
  fail "broken copy did not fail as expected (exit=$BROKEN_STATUS)"
fi
rm -rf "$SCRATCH"

echo ""
echo "8. The contract catches an out-of-order pipeline every time, not just once"
CONTRACT_CHECK=$("$PYTHON" - <<'PYEOF'
import sys
sys.path.insert(0, "examples")
import workflow_lib as w

caught = 0
for seed in range(5):
    try:
        w.run_pipeline(w.leaky_stages(), w.starting_artifact(seed=seed), enforce_contracts=True)
    except w.StageContractError:
        caught += 1
if caught != 5:
    print(f"ERROR: the contract caught {caught} of 5 out-of-order pipelines")
else:
    print("contract caught all five")
PYEOF
)
if [ "$CONTRACT_CHECK" = "contract caught all five" ]; then
  ok "the stage contract rejects the out-of-order pipeline at every seed tried"
else
  fail "contract check failed: $CONTRACT_CHECK"
fi

echo ""
echo "9. Offline, and nothing left behind"
if ! grep -rInE "https?://" examples/*.py starter/*.py > /dev/null 2>&1; then
  ok "no URLs inside examples/ or starter/ source -- this lab reaches no network"
else
  fail "found a URL inside examples/ or starter/"
fi
if [ -z "$(find . -path ./.venv -prune -o -type d -name '__pycache__' -print 2>/dev/null)" ]; then
  ok "no __pycache__ left behind"
else
  find . -path ./.venv -prune -o -type d -name '__pycache__' -exec rm -rf -- {} + 2>/dev/null
  ok "no __pycache__ left behind (cleaned during this run)"
fi
if [ ! -d .pytest_cache ]; then
  ok "no .pytest_cache left behind"
else
  rm -rf .pytest_cache
  ok "no .pytest_cache left behind (cleaned during this run)"
fi

echo ""
echo "---------------------------------------------------------------"
echo "$CHECKS checks, $FAILURES failure(s)"
if [ "$FAILURES" -ne 0 ]; then
  exit 1
fi
exit 0