diff --git a/conformance/README.md b/conformance/README.md
index eb94ad2..307cd0f 100644
--- a/conformance/README.md
+++ b/conformance/README.md
@@ -7,6 +7,11 @@ One set of language-neutral cases, a runner, and **one judge** (`run.py`).
 The judge is itself a codeupipe pipeline, in [judging/](judging/): one step
 reads the cases, one runs each target, one compares, and one writes the
 matrix. Each step is tested alone in `tests/conformance/test_judge_pipeline.py`.
+
+The Python runner is built the same way, in
+[runners/python/case_runner/](runners/python/case_runner/): one pipeline plays
+a case and one times a scenario. Its steps are tested alone in
+`tests/conformance/test_case_runner.py`.
 The dynamic run and each compiled form are held to exactly the same
 comparison.
 
diff --git a/conformance/runners/python/case_runner/__init__.py b/conformance/runners/python/case_runner/__init__.py
new file mode 100644
index 0000000..a453e76
--- /dev/null
+++ b/conformance/runners/python/case_runner/__init__.py
@@ -0,0 +1,6 @@
+"""The Python conformance runner, as codeupipe pipelines.
+
+``run_case_pipeline.py`` plays one case on one target and reports what
+happened. ``bench_scenario_pipeline.py`` times one scenario. A runner never
+judges: ``conformance/judging`` does.
+"""
diff --git a/conformance/runners/python/case_runner/attach_noop_hook_filter.py b/conformance/runners/python/case_runner/attach_noop_hook_filter.py
new file mode 100644
index 0000000..8170d41
--- /dev/null
+++ b/conformance/runners/python/case_runner/attach_noop_hook_filter.py
@@ -0,0 +1,19 @@
+"""AttachNoopHookFilter: a Hook that does nothing, for a scenario that measures the cost of Hooks."""
+
+import cup_ops as ops
+from codeupipe import Payload
+
+
+class AttachNoopHookFilter:
+    """Filter (sync).
+
+    Input keys: ``case`` (a scenario), ``pipeline``.
+    Output keys: ``hooks``: a list with one NoopHook for a scenario with
+    ``"hooks": "noop"``, or None. The pipeline gets its own NoopHook.
+    """
+
+    def call(self, payload: Payload) -> Payload:
+        if payload.get("case").get("hooks") != "noop":
+            return payload.insert("hooks", None)
+        payload.get("pipeline").use_hook(ops.NoopHook())
+        return payload.insert("hooks", [ops.NoopHook()])
diff --git a/conformance/runners/python/case_runner/attach_record_hook_filter.py b/conformance/runners/python/case_runner/attach_record_hook_filter.py
new file mode 100644
index 0000000..314be3b
--- /dev/null
+++ b/conformance/runners/python/case_runner/attach_record_hook_filter.py
@@ -0,0 +1,22 @@
+"""AttachRecordHookFilter: the Hook that records the lifecycle, for a case that asks for it."""
+
+import cup_ops as ops
+from codeupipe import Payload
+
+
+class AttachRecordHookFilter:
+    """Filter (sync).
+
+    Input keys: ``case``, ``pipeline``, ``compiled``, ``observe``.
+    Output keys: ``hooks``: the Hooks of the run, or None. A case with
+    ``"hooks"`` gets a RecordHook, on the pipeline too. Compiled code calls
+    Hooks only when it was compiled with ``observe``.
+    """
+
+    def call(self, payload: Payload) -> Payload:
+        wanted = payload.get("case").get("hooks") and (not payload.get("compiled") or payload.get("observe"))
+        if not wanted:
+            return payload.insert("hooks", None)
+        hook = ops.RecordHook()
+        payload.get("pipeline").use_hook(hook)
+        return payload.insert("hooks", [hook])
diff --git a/conformance/runners/python/case_runner/bench_each_scenario_filter.py b/conformance/runners/python/case_runner/bench_each_scenario_filter.py
new file mode 100644
index 0000000..f176275
--- /dev/null
+++ b/conformance/runners/python/case_runner/bench_each_scenario_filter.py
@@ -0,0 +1,38 @@
+"""BenchEachScenarioFilter: time each scenario, one after the other."""
+
+import pathlib
+import tempfile
+
+from codeupipe import Payload
+
+from .unsupported import ExportRefused
+
+
+class BenchEachScenarioFilter:
+    """Filter (sync).
+
+    Input keys: ``cases`` (the scenarios), ``target``.
+    Output keys: ``results``: for each scenario ``id``, ``us_per_run``,
+    ``relative`` and ``output``, or ``{"id", "unsupported": reason}`` when
+    the compiler refuses its pipeline.
+
+    ``each`` is the pipeline that times one scenario.
+    """
+
+    def __init__(self, each):
+        self.each = each
+
+    def call(self, payload: Payload) -> Payload:
+        target = payload.get("target")
+        results = [self.one(case, target, index) for index, case in enumerate(payload.get("cases"))]
+        return payload.insert("results", results)
+
+    def one(self, case, target, index) -> dict:
+        workdir = pathlib.Path(tempfile.mkdtemp(prefix="cup_bench_"))
+        try:
+            out = self.each.run(Payload({"case": case, "target": target, "workdir": workdir,
+                                         "module_name": f"bench_predicates_{index}"}))
+        except ExportRefused as reason:
+            return {"id": case["id"], "unsupported": str(reason)}
+        return {"id": case["id"], "us_per_run": out.get("us_per_run"), "relative": out.get("relative"),
+                "output": out.get("output")}
diff --git a/conformance/runners/python/case_runner/bench_scenario_pipeline.py b/conformance/runners/python/case_runner/bench_scenario_pipeline.py
new file mode 100644
index 0000000..fd8a41a
--- /dev/null
+++ b/conformance/runners/python/case_runner/bench_scenario_pipeline.py
@@ -0,0 +1,30 @@
+"""BenchScenarioPipeline: time one scenario on one target."""
+
+from codeupipe import Pipeline
+
+from .attach_noop_hook_filter import AttachNoopHookFilter
+from .build_pipeline_filter import BuildPipelineFilter
+from .export_filter import ExportFilter
+from .is_compiled_predicate import IsCompiledPredicate
+from .make_run_function_filter import MakeRunFunctionFilter
+from .measure_filter import MeasureFilter
+from .parse_target_filter import ParseTargetFilter
+from .write_predicates_filter import WritePredicatesFilter
+
+TARGETS = ("run", "static-steps", "static-inline", "static-observe")
+
+
+class BenchScenarioPipeline(Pipeline):
+    """Input keys: ``case`` (a scenario), ``target``, ``workdir``, ``module_name``.
+    Output keys: ``us_per_run``, ``relative``, ``output``.
+    """
+
+    def __init__(self):
+        super().__init__()
+        self.add_filter(ParseTargetFilter(TARGETS, "bench target"))
+        self.add_filter(WritePredicatesFilter())
+        self.add_filter(BuildPipelineFilter())
+        self.add_filter(AttachNoopHookFilter())
+        self.add_valve(IsCompiledPredicate(), ExportFilter())
+        self.add_filter(MakeRunFunctionFilter())
+        self.add_filter(MeasureFilter())
diff --git a/conformance/runners/python/case_runner/build_pipeline_filter.py b/conformance/runners/python/case_runner/build_pipeline_filter.py
new file mode 100644
index 0000000..d62bb4a
--- /dev/null
+++ b/conformance/runners/python/case_runner/build_pipeline_filter.py
@@ -0,0 +1,79 @@
+"""BuildPipelineFilter: the pipeline of a case, made from its steps."""
+
+import cup_ops as ops
+from codeupipe import Payload, Pipeline
+
+from .unsupported import Unsupported
+
+# The reference vocabulary (conformance/README.md): for each op, how its step is made from the case.
+_FILTERS = {
+    "set": lambda spec: ops.SetFilter(spec["key"], spec["value"]),
+    "copy": lambda spec: ops.CopyFilter(spec["from"], spec["to"]),
+    "add": lambda spec: ops.AddFilter(spec["key"], spec["by"]),
+    "append": lambda spec: ops.AppendFilter(spec["key"], spec["value"]),
+    "remove": lambda spec: ops.RemoveFilter(spec["key"]),
+    "merge": lambda spec: ops.MergeFilter(spec["value"]),
+    "has": lambda spec: ops.HasFilter(spec["key"], spec["to"]),
+    "mutate": lambda spec: ops.MutateFilter(spec["key"], spec["value"]),
+    "fail": lambda spec: ops.FailFilter(spec["message"]),
+    "run_again": lambda spec: ops.RunAgainFilter(spec["input"]),
+}
+_ASYNC_FILTERS = {**_FILTERS, "run_again": lambda spec: ops.AsyncRunAgainFilter(spec["input"])}
+_TAPS = {
+    "record": lambda spec: ops.RecordTap(spec["key"]),
+    "fail": lambda spec: ops.FailTap(spec["message"]),
+    "noop": lambda spec: ops.NoopTap(),
+}
+
+
+class BuildPipelineFilter:
+    """Filter (sync).
+
+    Input keys: ``case``, ``predicates`` (from WritePredicatesFilter),
+    ``run_async``.
+    Output keys: ``pipeline``, and ``label``: for each step label that the
+    pipeline reports, the name that the case gives that step. Steps are
+    reported by label (conformance/API.md) and the cases name them, and this
+    Filter built each step, so it knows which is which. A label that is not
+    in the map stays as it is, and fails the comparison.
+    """
+
+    def call(self, payload: Payload) -> Payload:
+        names = {}
+        pipeline = self.build(payload.get("case")["steps"], Pipeline, payload.get("predicates"),
+                              names, payload.get("run_async"))
+        label = {step_label: names[id(step)] for step_label, step, _type in pipeline._steps}
+        return payload.insert("pipeline", pipeline).insert("label", label)
+
+    def build(self, steps, cls, predicates, names, run_async=False):
+        pipeline = cls()
+        for step in steps:
+            kind, name = step["kind"], step["name"]
+            if kind == "filter":
+                made = self.filter(step, run_async)
+                pipeline.add_filter(made)
+            elif kind == "tap":
+                made = self.tap(step)
+                pipeline.add_tap(made)
+            elif kind == "valve":
+                pipeline.add_valve(getattr(predicates, f"when_{name}"), self.filter(step["filter"]))
+                made = pipeline._steps[-1][1]  # the Valve that the pipeline made: the step that the label names
+            elif kind == "pipeline":
+                made = self.build(step["steps"], ops.NestedPipeline, predicates, names)
+                pipeline.add_pipeline(made)
+            else:
+                raise Unsupported(f"unknown step kind {kind!r}")
+            names[id(made)] = name
+        return pipeline
+
+    def filter(self, spec, run_async=False):
+        make = (_ASYNC_FILTERS if run_async else _FILTERS).get(spec["op"])
+        if make is None:
+            raise Unsupported(f"unknown filter op {spec['op']!r}")
+        return make(spec)
+
+    def tap(self, spec):
+        make = _TAPS.get(spec["op"])
+        if make is None:
+            raise Unsupported(f"unknown tap op {spec['op']!r}")
+        return make(spec)
diff --git a/conformance/runners/python/case_runner/export_filter.py b/conformance/runners/python/case_runner/export_filter.py
new file mode 100644
index 0000000..54b04ca
--- /dev/null
+++ b/conformance/runners/python/case_runner/export_filter.py
@@ -0,0 +1,48 @@
+"""ExportFilter: compile the pipeline of the case and import the package from disk, as a customer would."""
+
+import importlib
+import sys
+
+from codeupipe import Payload
+
+from .unsupported import ExportRefused
+
+
+class ExportFilter:
+    """Filter (sync).
+
+    Input keys: ``pipeline``, ``lowering``, ``observe``, ``workdir``.
+    Output keys: ``package``: the compiled code, with its ``run`` function.
+
+    The compiler refuses a pipeline that it cannot keep equal. For the fast
+    forms that is an ExportRefused: the target cannot express the case. With
+    ``observe`` the error is given as it is.
+    """
+
+    def call(self, payload: Payload) -> Payload:
+        try:
+            package = self.load(payload.get("pipeline"), payload.get("lowering"), payload.get("observe"),
+                                payload.get("workdir"))
+        except Exception as error:  # noqa: BLE001 — the compiler refused the shape (for example Hooks)
+            if payload.get("observe"):
+                raise
+            raise ExportRefused(f"{type(error).__name__}: {error}") from error
+        return payload.insert("package", package)
+
+    def load(self, pipeline, lowering, observe, workdir):
+        from codeupipe.compile import build_export_pipeline_v2
+
+        out = build_export_pipeline_v2(write=False).run(
+            Payload({"pipeline": pipeline, "pattern": "flat", "lowering": lowering, "observe": observe}))
+        for file in out.get("files") or []:
+            target = workdir / file["path"]
+            target.parent.mkdir(parents=True, exist_ok=True)
+            target.write_text(file["content"])
+        package_parent = str(workdir / "src")
+        for name in [m for m in sys.modules if m == "steps" or m.startswith("steps.")]:
+            del sys.modules[name]
+        sys.path.insert(0, package_parent)
+        try:
+            return importlib.import_module("steps")
+        finally:
+            sys.path.remove(package_parent)
diff --git a/conformance/runners/python/case_runner/finish_result_filter.py b/conformance/runners/python/case_runner/finish_result_filter.py
new file mode 100644
index 0000000..6f0c991
--- /dev/null
+++ b/conformance/runners/python/case_runner/finish_result_filter.py
@@ -0,0 +1,16 @@
+"""FinishResultFilter: the id of the case and what its Taps recorded."""
+
+import conformance_sink as sink
+from codeupipe import Payload
+
+
+class FinishResultFilter:
+    """Filter (sync).
+
+    Input keys: ``case``, ``result``.
+    Output keys: ``result``, with ``id`` and ``taps``.
+    """
+
+    def call(self, payload: Payload) -> Payload:
+        result = {**payload.get("result"), "id": payload.get("case")["id"], "taps": list(sink.TAPS)}
+        return payload.insert("result", result)
diff --git a/conformance/runners/python/case_runner/is_compiled_predicate.py b/conformance/runners/python/case_runner/is_compiled_predicate.py
new file mode 100644
index 0000000..be83eca
--- /dev/null
+++ b/conformance/runners/python/case_runner/is_compiled_predicate.py
@@ -0,0 +1,13 @@
+"""IsCompiledPredicate: does the case run as compiled code?"""
+
+from codeupipe import Payload
+
+
+class IsCompiledPredicate:
+    """True when ``compiled`` is the value that was given (True unless said)."""
+
+    def __init__(self, compiled=True):
+        self.compiled = compiled
+
+    def test(self, payload: Payload) -> bool:
+        return payload.get("compiled") is self.compiled
diff --git a/conformance/runners/python/case_runner/make_run_function_filter.py b/conformance/runners/python/case_runner/make_run_function_filter.py
new file mode 100644
index 0000000..284d34e
--- /dev/null
+++ b/conformance/runners/python/case_runner/make_run_function_filter.py
@@ -0,0 +1,33 @@
+"""MakeRunFunctionFilter: the one function that a measurement calls again and again."""
+
+from codeupipe import Payload
+
+
+class MakeRunFunctionFilter:
+    """Filter (sync).
+
+    Input keys: ``case`` (a scenario), ``compiled``, ``observe``, ``hooks``,
+    and ``pipeline`` or ``package``.
+    Output keys: ``fn``: a function with no arguments that does one run of
+    the scenario on a new copy of its input. ``output``: what one run gives.
+    """
+
+    def call(self, payload: Payload) -> Payload:
+        fn = self.make(payload)
+        return payload.insert("fn", fn).insert("output", self.output(fn(), payload.get("compiled")))
+
+    def make(self, payload: Payload):
+        data = payload.get("case")["input"]
+        if not payload.get("compiled"):
+            pipeline = payload.get("pipeline")
+            return lambda: pipeline.run(Payload(dict(data)))
+        run = payload.get("package").run
+        if not payload.get("observe"):
+            return lambda: run(dict(data))
+        hooks = payload.get("hooks")
+        state = {}
+        return lambda: run(dict(data), state=state, hooks=hooks)
+
+    def output(self, out, compiled):
+        """The dynamic run gives a Payload, and compiled code gives a dict."""
+        return out if compiled else out.to_dict()
diff --git a/conformance/runners/python/case_runner/measure_filter.py b/conformance/runners/python/case_runner/measure_filter.py
new file mode 100644
index 0000000..092fa5e
--- /dev/null
+++ b/conformance/runners/python/case_runner/measure_filter.py
@@ -0,0 +1,61 @@
+"""MeasureFilter: how long one run takes."""
+
+import statistics
+import time
+
+from codeupipe import Payload
+
+WARMUP = 20000
+RUNS = 20000
+REPEATS = 5
+
+_REFERENCE_KEYS = tuple("k%d" % i for i in range(10))
+
+
+def _reference_write(data, key):
+    data[key] = data.get(key, 0) + 1
+    return data
+
+
+class MeasureFilter:
+    """Filter (sync).
+
+    Input keys: ``fn`` (from MakeRunFunctionFilter).
+    Output keys: ``us_per_run`` (µs for one run: the best of REPEATS x RUNS)
+    and ``relative`` (the time of ``fn`` as a ratio to a reference workload).
+
+    The relative time is for comparing runs on different machines and under
+    different load: each repeat times the reference workload and then
+    ``fn``, back to back, so whatever slows one slows the other, and the
+    median of the ratios of the repeats is reported.
+    """
+
+    def call(self, payload: Payload) -> Payload:
+        us_per_run, relative = self.measure(payload.get("fn"))
+        return payload.insert("us_per_run", us_per_run).insert("relative", relative)
+
+    def measure(self, fn):
+        for _ in range(WARMUP):
+            fn()
+            self.reference()
+        best, ratios = None, []
+        for _ in range(REPEATS):
+            reference = self.timed(self.reference)
+            elapsed = self.timed(fn)
+            ratios.append(elapsed / reference)
+            best = elapsed if best is None else min(best, elapsed)
+        return best / RUNS * 1e6, statistics.median(ratios)
+
+    def timed(self, fn) -> float:
+        started = time.perf_counter()
+        for _ in range(RUNS):
+            fn()
+        return time.perf_counter() - started
+
+    @staticmethod
+    def reference():
+        """The reference workload: ten dict writes behind ten function calls, with no framework."""
+        data = {"a": 1}
+        for key in _REFERENCE_KEYS:
+            data = _reference_write(data, key)
+        return data
diff --git a/conformance/runners/python/case_runner/name_steps_filter.py b/conformance/runners/python/case_runner/name_steps_filter.py
new file mode 100644
index 0000000..5661b5a
--- /dev/null
+++ b/conformance/runners/python/case_runner/name_steps_filter.py
@@ -0,0 +1,35 @@
+"""NameStepsFilter: give each step in the result the name that the case gives it."""
+
+from codeupipe import Payload
+
+
+class NameStepsFilter:
+    """Filter (sync).
+
+    Input keys: ``result``, ``label`` ({step label: name in the case}).
+    Output keys: ``result``, with the name of the case for each step in
+    ``state``, ``pipeline_state``, ``inner_states`` and ``hooks``. A label
+    that is not in ``label`` stays as it is.
+    """
+
+    def call(self, payload: Payload) -> Payload:
+        label = payload.get("label")
+        named = {key: self.named(key, value, label) for key, value in payload.get("result").items()}
+        return payload.insert("result", named)
+
+    def named(self, key, value, label):
+        """The value of one key of the result, with the names of the case where it has steps."""
+        if key in ("state", "pipeline_state"):
+            return self.state(value, label)
+        if key == "inner_states":
+            return [self.state(one, label) for one in value]
+        if key == "hooks":
+            return [self.event(event, label) for event in value]
+        return value
+
+    def state(self, state, label) -> dict:
+        return {key: [label.get(step, step) for step in steps] for key, steps in state.items()}
+
+    def event(self, event, label) -> str:
+        phase, _, step = event.partition(":")
+        return f"{phase}:{label.get(step, step)}"
diff --git a/conformance/runners/python/case_runner/parse_target_filter.py b/conformance/runners/python/case_runner/parse_target_filter.py
new file mode 100644
index 0000000..fab6c17
--- /dev/null
+++ b/conformance/runners/python/case_runner/parse_target_filter.py
@@ -0,0 +1,36 @@
+"""ParseTargetFilter: what the name of a target means."""
+
+from codeupipe import Payload
+
+_DYNAMIC = {"compiled": False, "run_async": False, "lowering": "", "observe": False}
+_TARGETS = {
+    "run": _DYNAMIC,
+    "run_async": {**_DYNAMIC, "run_async": True},
+    "static-steps": {"compiled": True, "run_async": False, "lowering": "steps", "observe": False},
+    "static-inline": {"compiled": True, "run_async": False, "lowering": "inline", "observe": False},
+    "static-observe": {"compiled": True, "run_async": False, "lowering": "steps", "observe": True},
+}
+
+
+class ParseTargetFilter:
+    """Filter (sync).
+
+    Input keys: ``target`` (``run``, ``run_async``, ``static-steps``,
+    ``static-inline`` or ``static-observe``).
+    Output keys: ``compiled`` (the case runs as compiled code), ``run_async``,
+    ``lowering`` (the layout of the compiled code) and ``observe`` (the
+    compiled code records State and calls Hooks).
+
+    ``allowed`` is the targets that this run takes, and ``what`` is the word
+    for them in the message for another name.
+    """
+
+    def __init__(self, allowed=tuple(_TARGETS), what="target"):
+        self.allowed = allowed
+        self.what = what
+
+    def call(self, payload: Payload) -> Payload:
+        target = payload.get("target")
+        if target not in self.allowed:
+            raise SystemExit(f"unknown {self.what} {target!r}")
+        return payload.merge(Payload(dict(_TARGETS[target])))
diff --git a/conformance/runners/python/case_runner/run_case_pipeline.py b/conformance/runners/python/case_runner/run_case_pipeline.py
new file mode 100644
index 0000000..6b72a81
--- /dev/null
+++ b/conformance/runners/python/case_runner/run_case_pipeline.py
@@ -0,0 +1,36 @@
+"""RunCasePipeline: play one conformance case on one target and report what happened."""
+
+from codeupipe import Pipeline
+
+from .attach_record_hook_filter import AttachRecordHookFilter
+from .build_pipeline_filter import BuildPipelineFilter
+from .export_filter import ExportFilter
+from .finish_result_filter import FinishResultFilter
+from .is_compiled_predicate import IsCompiledPredicate
+from .name_steps_filter import NameStepsFilter
+from .parse_target_filter import ParseTargetFilter
+from .run_compiled_filter import RunCompiledFilter
+from .run_dynamic_filter import RunDynamicFilter
+from .write_predicates_filter import WritePredicatesFilter
+
+
+class RunCasePipeline(Pipeline):
+    """Input keys: ``case``, ``target``, ``workdir``, ``module_name``.
+    Output keys: ``result`` (conformance/README.md, "Runner protocol").
+
+    The runner never judges. A step that raises Unsupported, or any other
+    error before the case runs, means that the target cannot express the
+    case: RunEachCaseFilter reports that.
+    """
+
+    def __init__(self):
+        super().__init__()
+        self.add_filter(ParseTargetFilter())
+        self.add_filter(WritePredicatesFilter())
+        self.add_filter(BuildPipelineFilter())
+        self.add_filter(AttachRecordHookFilter())
+        self.add_valve(IsCompiledPredicate(), ExportFilter())
+        self.add_valve(IsCompiledPredicate(False), RunDynamicFilter())
+        self.add_valve(IsCompiledPredicate(), RunCompiledFilter())
+        self.add_filter(NameStepsFilter())
+        self.add_filter(FinishResultFilter())
diff --git a/conformance/runners/python/case_runner/run_compiled_filter.py b/conformance/runners/python/case_runner/run_compiled_filter.py
new file mode 100644
index 0000000..ebef4d5
--- /dev/null
+++ b/conformance/runners/python/case_runner/run_compiled_filter.py
@@ -0,0 +1,55 @@
+"""RunCompiledFilter: run the case as compiled code."""
+
+import copy
+
+import conformance_sink as sink
+from codeupipe import Payload
+
+
+class RunCompiledFilter:
+    """Filter (sync).
+
+    Input keys: ``case``, ``package`` (from ExportFilter), ``observe``,
+    ``hooks``.
+    Output keys: ``result``: ``output`` or ``error``, and ``input_after``.
+    With ``observe`` also ``state``, ``inner_states`` and, for a case that
+    asks for them, ``hooks``. Compiled code is a function, so it has no
+    ``pipeline_state``. The steps in the result are labels: NameStepsFilter
+    gives them the names of the case.
+    """
+
+    def call(self, payload: Payload) -> Payload:
+        case, package, observe = payload.get("case"), payload.get("package"), payload.get("observe")
+        hooks = payload.get("hooks")
+        data = copy.deepcopy(case["input"])
+        result = {}
+        state = {}
+        inner_states = []
+
+        def again(inner):
+            # The compiled form is a function. To run the pipeline again is to call that function again,
+            # and the caller gives each call its own State.
+            if not observe:
+                return package.run(inner)
+            inner_states.append({})
+            return package.run(inner, state=inner_states[-1], hooks=None)
+
+        sink.RUN_AGAIN[0] = again
+        try:
+            if observe:
+                result["output"] = package.run(data, state=state, hooks=hooks)
+            else:
+                result["output"] = package.run(data)
+        except Exception as error:  # noqa: BLE001
+            result["error"] = f"{type(error).__name__}: {error}"
+        if observe:
+            result["state"] = self.report(state)
+            result["inner_states"] = [self.report(one) for one in inner_states]
+            if case.get("hooks"):
+                result["hooks"] = list(sink.HOOKS)
+        result["input_after"] = data
+        return payload.insert("result", result)
+
+    def report(self, state) -> dict:
+        return {"executed": list(state["executed"]), "skipped": list(state["skipped"]),
+                "errors": [step for step, _error in state["errors"]]}
diff --git a/conformance/runners/python/case_runner/run_dynamic_filter.py b/conformance/runners/python/case_runner/run_dynamic_filter.py
new file mode 100644
index 0000000..8bd4d0b
--- /dev/null
+++ b/conformance/runners/python/case_runner/run_dynamic_filter.py
@@ -0,0 +1,56 @@
+"""RunDynamicFilter: run the case with ``Pipeline.run`` or ``Pipeline.run_async``."""
+
+import asyncio
+import copy
+
+import conformance_sink as sink
+from codeupipe import Payload, State
+
+
+class RunDynamicFilter:
+    """Filter (sync).
+
+    Input keys: ``case``, ``pipeline``, ``run_async``, ``hooks``.
+    Output keys: ``result``: ``output`` or ``error``, ``state`` (the State of
+    the run), ``pipeline_state`` (``state`` on the Pipeline after the run),
+    ``inner_states`` (the State of each run that a run_again step started),
+    ``hooks`` for a case that asks for them, and ``input_after``. The steps
+    in it are labels: NameStepsFilter gives them the names of the case.
+    """
+
+    def call(self, payload: Payload) -> Payload:
+        case, pipeline, run_async = payload.get("case"), payload.get("pipeline"), payload.get("run_async")
+        data = copy.deepcopy(case["input"])
+        # With "given_state", the caller gives the State of the run. If not, the run makes one.
+        given = [State()] if case.get("given_state") else []
+        result = {}
+
+        def again(inner):
+            state = State()
+            sink.INNER_STATES.append(state)
+            return pipeline.run_async(Payload(inner), state) if run_async else pipeline.run(Payload(inner), state)
+
+        sink.RUN_AGAIN[0] = again
+        try:
+            if run_async:
+                out = asyncio.run(pipeline.run_async(Payload(data), *given))
+            else:
+                out = pipeline.run(Payload(data), *given)
+            result["output"] = out.to_dict()
+        except Exception as error:  # noqa: BLE001 — the error is the result
+            result["error"] = f"{type(error).__name__}: {error}"
+
+        # The State that these two lines read is that of the pipeline of the case, which this step ran.
+        # It is what the case asks for, not the State of the run that this step is in.
+        ran = pipeline.state  # cup: allow-state-read the State of the pipeline of the case is the result that this step reports
+        result["state"] = self.report(given[0] if given else ran)
+        result["pipeline_state"] = self.report(ran)
+        result["inner_states"] = [self.report(state) for state in sink.INNER_STATES]
+        if payload.get("hooks"):
+            result["hooks"] = list(sink.HOOKS)
+        result["input_after"] = data
+        return payload.insert("result", result)
+
+    def report(self, state) -> dict:
+        return {"executed": list(state.executed), "skipped": list(state.skipped),
+                "errors": [step for step, _error in state.errors]}
diff --git a/conformance/runners/python/case_runner/run_each_case_filter.py b/conformance/runners/python/case_runner/run_each_case_filter.py
new file mode 100644
index 0000000..fd3d8e0
--- /dev/null
+++ b/conformance/runners/python/case_runner/run_each_case_filter.py
@@ -0,0 +1,40 @@
+"""RunEachCaseFilter: play each case, one after the other, each in its own directory."""
+
+import pathlib
+import tempfile
+
+import conformance_sink as sink
+from codeupipe import Payload
+
+from .unsupported import Unsupported
+
+
+class RunEachCaseFilter:
+    """Filter (sync).
+
+    Input keys: ``cases``, ``target``.
+    Output keys: ``results``: for each case its result, or ``{"id",
+    "unsupported": reason}`` when the target cannot express it.
+
+    ``each`` is the pipeline that plays one case.
+    """
+
+    def __init__(self, each):
+        self.each = each
+
+    def call(self, payload: Payload) -> Payload:
+        target = payload.get("target")
+        results = [self.one(case, target, index) for index, case in enumerate(payload.get("cases"))]
+        return payload.insert("results", results)
+
+    def one(self, case, target, index) -> dict:
+        sink.reset()
+        workdir = pathlib.Path(tempfile.mkdtemp(prefix="cup_conf_"))
+        try:
+            out = self.each.run(Payload({"case": case, "target": target, "workdir": workdir,
+                                         "module_name": f"conf_predicates_{index}"}))
+        except Unsupported as reason:
+            return {"id": case["id"], "unsupported": str(reason)}
+        except Exception as error:  # noqa: BLE001 — for example the compiler refuses a shape
+            return {"id": case["id"], "unsupported": f"{type(error).__name__}: {error}"}
+        return out.get("result")
diff --git a/conformance/runners/python/case_runner/runner_pipeline.py b/conformance/runners/python/case_runner/runner_pipeline.py
new file mode 100644
index 0000000..0ca304c
--- /dev/null
+++ b/conformance/runners/python/case_runner/runner_pipeline.py
@@ -0,0 +1,15 @@
+"""RunnerPipeline: the whole runner for one call: each case, or each scenario, on one target."""
+
+from codeupipe import Pipeline
+
+
+class RunnerPipeline(Pipeline):
+    """Input keys: ``cases``, ``target``. Output keys: ``results``.
+
+    ``each`` is the Filter that goes through the cases: RunEachCaseFilter to
+    play them, or BenchEachScenarioFilter to time them.
+    """
+
+    def __init__(self, each):
+        super().__init__()
+        self.add_filter(each)
diff --git a/conformance/runners/python/case_runner/unsupported.py b/conformance/runners/python/case_runner/unsupported.py
new file mode 100644
index 0000000..0d2cb73
--- /dev/null
+++ b/conformance/runners/python/case_runner/unsupported.py
@@ -0,0 +1,9 @@
+"""The errors that say "this target cannot express this case"."""
+
+
+class Unsupported(Exception):
+    """The target cannot express the case. The message is the reason."""
+
+
+class ExportRefused(Unsupported):
+    """The compiler refused the pipeline of the case."""
diff --git a/conformance/runners/python/case_runner/write_predicates_filter.py b/conformance/runners/python/case_runner/write_predicates_filter.py
new file mode 100644
index 0000000..09145ff
--- /dev/null
+++ b/conformance/runners/python/case_runner/write_predicates_filter.py
@@ -0,0 +1,55 @@
+"""WritePredicatesFilter: the Predicates of a case, as real classes in a module on disk."""
+
+import importlib
+import json
+import sys
+
+from codeupipe import Payload
+
+from .unsupported import Unsupported
+
+
+class WritePredicatesFilter:
+    """Filter (sync).
+
+    Input keys: ``case``, ``workdir`` (a directory for this case),
+    ``module_name``.
+    Output keys: ``predicates``: a module that has, for each Valve of the
+    case, an instance ``when_<name>`` of a Predicate class.
+
+    The Predicates are written as source code and imported, because the
+    compiler reads the source of a Predicate and refuses anything else.
+    """
+
+    def call(self, payload: Payload) -> Payload:
+        workdir, module_name = payload.get("workdir"), payload.get("module_name")
+        sources = self.sources(payload.get("case")["steps"])
+        (workdir / f"{module_name}.py").write_text("\n\n".join(sources) or "\n")
+        sys.path.insert(0, str(workdir))
+        try:
+            return payload.insert("predicates", importlib.import_module(module_name))
+        finally:
+            sys.path.remove(str(workdir))
+
+    def sources(self, steps) -> list:
+        """The source of the Predicate of each Valve, with those of the nested pipelines."""
+        found = []
+        for step in steps:
+            if step["kind"] == "valve":
+                found.append(self.source(step["name"], step["when"]))
+            elif step["kind"] == "pipeline":
+                found.extend(self.sources(step["steps"]))
+        return found
+
+    def source(self, name, when) -> str:
+        key = json.dumps(when["key"])
+        if when["op"] == "equals":
+            body = f"payload.get({key}) == {when['value']!r}"
+        elif when["op"] == "truthy":
+            body = f"bool(payload.get({key}))"
+        else:
+            raise Unsupported(f"unknown valve op {when['op']!r}")
+        # a predicate object (conformance/API.md, "Valve"): a class with test(),
+        # plus a module-level instance under the name the builders look up
+        return (f"class When_{name}:\n    def test(self, payload):\n        return {body}\n\n\n"
+                f"when_{name} = When_{name}()\n")
diff --git a/conformance/runners/python/runner.py b/conformance/runners/python/runner.py
index d450a58..77d5d5a 100644
--- a/conformance/runners/python/runner.py
+++ b/conformance/runners/python/runner.py
@@ -9,391 +9,45 @@
     static-inline  codeupipe.compile export, lowering="inline", loaded from disk
     static-observe codeupipe.compile export, observe=True: State and Hooks too
 
+With ``bench:<target>`` it times the scenarios that it gets, and does not
+play cases (conformance/bench.py).
+
+The runner is two codeupipe pipelines, in ``case_runner/``: one plays a
+case, and one times a scenario. This file is its command line.
+
 The runner never judges; conformance/run.py compares every target's results
 against the same expectations. The fast static targets have no State or
 Hooks, so they omit those fields and the orchestrator skips those checks;
 static-observe reports both and is judged on them.
 """
 
-import asyncio
-import copy
-import importlib
 import json
 import pathlib
 import sys
-import tempfile
 
 HERE = pathlib.Path(__file__).resolve().parent
 sys.path.insert(0, str(HERE))
 
-import conformance_sink as sink  # noqa: E402
-import cup_ops as ops  # noqa: E402
-from codeupipe import Payload, Pipeline, State  # noqa: E402
-  # noqa: E402
-
-
-class Unsupported(Exception):
-    pass
-
-
-# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges
-def _filter(spec, run_async=False):
-    op = spec["op"]
-    if op == "run_again":
-        return (ops.AsyncRunAgainFilter if run_async else ops.RunAgainFilter)(spec["input"])
-    if op == "set":
-        return ops.SetFilter(spec["key"], spec["value"])
-    if op == "copy":
-        return ops.CopyFilter(spec["from"], spec["to"])
-    if op == "add":
-        return ops.AddFilter(spec["key"], spec["by"])
-    if op == "append":
-        return ops.AppendFilter(spec["key"], spec["value"])
-    if op == "remove":
-        return ops.RemoveFilter(spec["key"])
-    if op == "merge":
-        return ops.MergeFilter(spec["value"])
-    if op == "has":
-        return ops.HasFilter(spec["key"], spec["to"])
-    if op == "mutate":
-        return ops.MutateFilter(spec["key"], spec["value"])
-    if op == "fail":
-        return ops.FailFilter(spec["message"])
-    raise Unsupported(f"unknown filter op {op!r}")
-
-
-# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges
-def _tap(spec):
-    if spec["op"] == "record":
-        return ops.RecordTap(spec["key"])
-    if spec["op"] == "fail":
-        return ops.FailTap(spec["message"])
-    if spec["op"] == "noop":
-        return ops.NoopTap()
-    raise Unsupported(f"unknown tap op {spec['op']!r}")
-
-
-# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges
-def _predicate_source(name, when):
-    """Valve predicates are generated as real predicate classes so static
-    lowering can recover their source (it refuses anything else)."""
-    key = json.dumps(when["key"])
-    if when["op"] == "equals":
-        body = f"payload.get({key}) == {when['value']!r}"
-    elif when["op"] == "truthy":
-        body = f"bool(payload.get({key}))"
-    else:
-        raise Unsupported(f"unknown valve op {when['op']!r}")
-    # a predicate object (conformance/API.md, "Valve"): a class with test(),
-    # plus a module-level instance under the name the builders look up
-    return (f"class When_{name}:\n    def test(self, payload):\n        return {body}\n\n\n"
-            f"when_{name} = When_{name}()\n")
-
-
-# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges
-def _predicates(steps, module_dir, module_name):
-    sources = []
-
-    def walk(items):
-        for step in items:
-            if step["kind"] == "valve":
-                sources.append(_predicate_source(step["name"], step["when"]))
-            elif step["kind"] == "pipeline":
-                walk(step["steps"])
-
-    walk(steps)
-    path = module_dir / f"{module_name}.py"
-    path.write_text("\n\n".join(sources) or "\n")
-    sys.path.insert(0, str(module_dir))
-    try:
-        return importlib.import_module(module_name)
-    finally:
-        sys.path.remove(str(module_dir))
-
-
-# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges
-def build(case, cls, predicates, names, run_async=False):
-    p = cls()
-    for step in case["steps"]:
-        kind, name = step["kind"], step["name"]
-        if kind == "filter":
-            obj = _filter(step, run_async)
-            p.add_filter(obj)
-        elif kind == "tap":
-            obj = _tap(step)
-            p.add_tap(obj)
-        elif kind == "valve":
-            p.add_valve(getattr(predicates, f"when_{name}"), _filter(step["filter"]))
-            obj = p._steps[-1][1]  # the Valve that the pipeline made: the step that labels() names
-        elif kind == "pipeline":
-            if not hasattr(p, "add_pipeline"):
-                raise Unsupported(f"{cls.__name__} has no add_pipeline")
-            obj = build({"steps": step["steps"]}, ops.NestedPipeline, predicates, names)
-            p.add_pipeline(obj)
-        else:
-            raise Unsupported(f"unknown step kind {kind!r}")
-        names[id(obj)] = name
-    return p
-
-
-def labels(p, names):
-    """label -> the case's step name, for the steps of ``p``.
-
-    Steps are reported by label (conformance/API.md), while the
-    shared cases name them. The runner built every step, so it knows which
-    case step each label belongs to. A label reported that is not in
-    this map (wrong, or inconsistent between modes) stays as is and fails
-    the comparison.
-    """
-    return {label: names[id(step)] for label, step, _type in p._steps}
-
-
-def _named(steps, label):
-    return [label.get(step, step) for step in steps]
-
-
-# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges
-def _named_hooks(events, label):
-    out = []
-    for event in events:
-        phase, _, step = event.partition(":")
-        out.append(f"{phase}:{label.get(step, step)}")
-    return out
-
-
-# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges
-def _dynamic(case, predicates, run_async=False):
-    names = {}
-    p = build(case, Pipeline, predicates, names, run_async)
-    label = labels(p, names)
-    if case.get("hooks"):
-        p.use_hook(ops.RecordHook())
-    data = copy.deepcopy(case["input"])
-    # With "given_state", the caller gives the State of the run. If not, the run makes one.
-    given = [State()] if case.get("given_state") else []
-    result = {}
-
-    def again(inner):
-        state = State()
-        sink.INNER_STATES.append(state)
-        return p.run_async(Payload(inner), state) if run_async else p.run(Payload(inner), state)
-
-    sink.RUN_AGAIN[0] = again
-    try:
-        if run_async:
-            out = asyncio.run(p.run_async(Payload(data), *given))
-        else:
-            out = p.run(Payload(data), *given)
-        result["output"] = out.to_dict()
-    except Exception as error:  # noqa: BLE001 — the error is the result
-        result["error"] = f"{type(error).__name__}: {error}"
-
-    def report(state):
-        return {
-            "executed": _named(state.executed, label),
-            "skipped": _named(state.skipped, label),
-            "errors": _named([step for step, _ in state.errors], label),
-        }
-
-    result["state"] = report(given[0] if given else p.state)
-    result["pipeline_state"] = report(p.state)
-    result["inner_states"] = [report(state) for state in sink.INNER_STATES]
-    if case.get("hooks"):
-        result["hooks"] = _named_hooks(sink.HOOKS, label)
-    result["input_after"] = data
-    return result
-
-
-# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges
-def _load_export(p, lowering, observe, workdir):
-    """Export ``p`` and import the package from disk, as a customer would."""
-    from codeupipe.compile import build_export_pipeline_v2
-
-    out = build_export_pipeline_v2(write=False).run(
-        Payload({"pipeline": p, "pattern": "flat", "lowering": lowering, "observe": observe}))
-    for f in out.get("files") or []:
-        target = workdir / f["path"]
-        target.parent.mkdir(parents=True, exist_ok=True)
-        target.write_text(f["content"])
-    package_parent = str(workdir / "src")
-    for name in [m for m in sys.modules if m == "steps" or m.startswith("steps.")]:
-        del sys.modules[name]
-    sys.path.insert(0, package_parent)
-    try:
-        return importlib.import_module("steps")
-    finally:
-        sys.path.remove(package_parent)
-
-
-# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges
-def _static(case, predicates, lowering, workdir, observe=False):
-    # The compiled form always gets its State from the caller, so "given_state" needs nothing more.
-    names = {}
-    p = build(case, Pipeline, predicates, names)
-    label = labels(p, names)
-    hooks = [ops.RecordHook()] if observe and case.get("hooks") else None
-    if hooks:
-        p.use_hook(hooks[0])
-    package = _load_export(p, lowering, observe, workdir)
-    data = copy.deepcopy(case["input"])
-    result = {}
-    state = {}
-    inner_states = []
-
-    def again(inner):
-        # The compiled form is a function. To run the pipeline again is to call that function again,
-        # and the caller gives each call its own State.
-        if not observe:
-            return package.run(inner)
-        inner_states.append({})
-        return package.run(inner, state=inner_states[-1], hooks=None)
-
-    sink.RUN_AGAIN[0] = again
-    try:
-        if observe:
-            result["output"] = package.run(data, state=state, hooks=hooks)
-        else:
-            result["output"] = package.run(data)
-    except Exception as error:  # noqa: BLE001
-        result["error"] = f"{type(error).__name__}: {error}"
-    if observe:
-        def report(one):
-            return {
-                "executed": _named(one["executed"], label),
-                "skipped": _named(one["skipped"], label),
-                "errors": _named([step for step, _ in one["errors"]], label),
-            }
-
-        result["state"] = report(state)
-        result["inner_states"] = [report(one) for one in inner_states]
-        if case.get("hooks"):
-            result["hooks"] = _named_hooks(sink.HOOKS, label)
-    result["input_after"] = data
-    return result
-
-
-# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges
-def run_case(case, target, index):
-    sink.reset()
-    tmp = pathlib.Path(tempfile.mkdtemp(prefix="cup_conf_"))
-    try:
-        predicates = _predicates(case["steps"], tmp, f"conf_predicates_{index}")
-        if target == "run":
-            result = _dynamic(case, predicates)
-        elif target == "run_async":
-            result = _dynamic(case, predicates, run_async=True)
-        elif target in ("static-steps", "static-inline"):
-            result = _static(case, predicates, target.split("-", 1)[1], tmp)
-        elif target == "static-observe":
-            result = _static(case, predicates, "steps", tmp, observe=True)
-        else:
-            raise SystemExit(f"unknown target {target!r}")
-    except Unsupported as reason:
-        return {"id": case["id"], "unsupported": str(reason)}
-    except Exception as error:  # noqa: BLE001 — e.g. export refusing a shape
-        return {"id": case["id"], "unsupported": f"{type(error).__name__}: {error}"}
-    result["id"] = case["id"]
-    result["taps"] = list(sink.TAPS)
-    return result
-
-
-BENCH_WARMUP = 20000
-BENCH_RUNS = 20000
-BENCH_REPEATS = 5
-
-
-_REFERENCE_KEYS = tuple("k%d" % i for i in range(10))
-
-
-def _reference_write(data, key):
-    data[key] = data.get(key, 0) + 1
-    return data
-
-
-# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges
-def _reference():
-    """The reference workload: ten dict writes behind ten function calls,
-    with no framework. A pipeline's time is reported relative to it."""
-    data = {"a": 1}
-    for key in _REFERENCE_KEYS:
-        data = _reference_write(data, key)
-    return data
-
-
-# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges
-def _timed(fn):
-    import time
-    t0 = time.perf_counter()
-    for _ in range(BENCH_RUNS):
-        fn()
-    return time.perf_counter() - t0
-
-
-# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges
-def _measure(fn):
-    """(µs per run, time relative to the reference workload).
-
-    µs per run is the best of BENCH_REPEATS x BENCH_RUNS. The relative time
-    is for comparing runs on different machines and under different load:
-    each repeat times the reference workload and then ``fn``, back to back,
-    so whatever slows one slows the other, and the median of the repeats'
-    ratios is reported.
-    """
-    import statistics
-    for _ in range(BENCH_WARMUP):
-        fn()
-        _reference()
-    best, ratios = None, []
-    for _ in range(BENCH_REPEATS):
-        reference = _timed(_reference)
-        elapsed = _timed(fn)
-        ratios.append(elapsed / reference)
-        best = elapsed if best is None else min(best, elapsed)
-    return best / BENCH_RUNS * 1e6, statistics.median(ratios)
+from codeupipe import Payload  # noqa: E402
 
+from case_runner.bench_each_scenario_filter import BenchEachScenarioFilter  # noqa: E402
+from case_runner.bench_scenario_pipeline import BenchScenarioPipeline  # noqa: E402
+from case_runner.run_case_pipeline import RunCasePipeline  # noqa: E402
+from case_runner.run_each_case_filter import RunEachCaseFilter  # noqa: E402
+from case_runner.runner_pipeline import RunnerPipeline  # noqa: E402
 
-# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges
-def bench(target, scenarios):
-    results = []
-    for index, case in enumerate(scenarios):
-        tmp = pathlib.Path(tempfile.mkdtemp(prefix="cup_bench_"))
-        predicates = _predicates(case["steps"], tmp, f"bench_predicates_{index}")
-        p = build(case, Pipeline, predicates, {})
-        if case.get("hooks") == "noop":
-            p.use_hook(ops.NoopHook())
-        data = case["input"]
-        if target == "run":
-            fn = lambda: p.run(Payload(dict(data)))  # noqa: E731
-            output = fn().to_dict()
-        elif target in ("static-inline", "static-steps"):
-            try:
-                run = _load_export(p, target.split("-", 1)[1], False, tmp).run
-            except Exception as error:  # noqa: BLE001 — export refused the shape (e.g. Hooks)
-                results.append({"id": case["id"], "unsupported": f"{type(error).__name__}: {error}"})
-                continue
-            fn = lambda: run(dict(data))  # noqa: E731
-            output = fn()
-        elif target == "static-observe":
-            run = _load_export(p, "steps", True, tmp).run
-            hooks = [ops.NoopHook()] if case.get("hooks") == "noop" else None
-            state = {}
-            fn = lambda: run(dict(data), state=state, hooks=hooks)  # noqa: E731
-            output = fn()
-        else:
-            raise SystemExit(f"unknown bench target {target!r}")
-        us_per_run, relative = _measure(fn)
-        results.append({"id": case["id"], "us_per_run": us_per_run, "relative": relative, "output": output})
-    return results
+# Is the first argument "bench:<target>"? Then the runner times scenarios. If not, it plays cases.
+_PIPELINES = {
+    False: lambda: RunnerPipeline(RunEachCaseFilter(RunCasePipeline())),
+    True: lambda: RunnerPipeline(BenchEachScenarioFilter(BenchScenarioPipeline())),
+}
 
 
-# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges
 def main():
     target = sys.argv[1]
-    if target.startswith("bench:"):
-        json.dump(bench(target.split(":", 1)[1], json.load(sys.stdin)), sys.stdout)
-        return
-    cases = json.load(sys.stdin)
-    json.dump([run_case(case, target, i) for i, case in enumerate(cases)], sys.stdout)
+    pipeline = _PIPELINES[target.startswith("bench:")]()
+    out = pipeline.run(Payload({"cases": json.load(sys.stdin), "target": target.removeprefix("bench:")}))
+    json.dump(out.get("results"), sys.stdout)
 
 
 if __name__ == "__main__":
diff --git a/scorecard/vocab.py b/scorecard/vocab.py
index a2790b5..606e47b 100644
--- a/scorecard/vocab.py
+++ b/scorecard/vocab.py
@@ -1,7 +1,7 @@
 """The conformance vocabulary, reused (not reimplemented) from the Python
-conformance runner: ``runner.build`` turns a case's steps into a live
-Pipeline, ``runner._predicate_source`` writes Valve predicates as real
-``def``s, and ``runner._load_export`` exports and imports the package from
+conformance runner: ``BuildPipelineFilter`` turns a case's steps into a live
+Pipeline, ``WritePredicatesFilter`` writes Valve predicates as real
+classes, and ``ExportFilter`` exports and imports the package from
 disk, as a customer would.
 
 The only thing done differently here: predicate ``def``s get index-based
@@ -27,11 +27,13 @@
 
 import conformance_sink as sink  # noqa: E402
 import cup_ops as ops  # noqa: E402
-import runner  # noqa: E402
+from case_runner.build_pipeline_filter import BuildPipelineFilter  # noqa: E402
+from case_runner.export_filter import ExportFilter  # noqa: E402
+from case_runner.write_predicates_filter import WritePredicatesFilter  # noqa: E402
 from codeupipe import Payload, Pipeline  # noqa: E402
 from codeupipe.compile.errors import UnsupportedFilterShape  # noqa: E402
 
-__all__ = ["sink", "ops", "runner", "Payload", "Pipeline", "UnsupportedFilterShape",
+__all__ = ["sink", "ops", "Payload", "Pipeline", "UnsupportedFilterShape",
            "make_predicates", "build_pipeline", "load_export", "new_workdir", "walk_steps"]
 
 _COUNTER = itertools.count()
@@ -55,7 +57,7 @@ def make_predicates(steps, workdir):
     """A namespace with ``when_<step name>`` for every Valve, each a real
     ``def`` in a real file (static lowering needs its source)."""
     valves = [s for s in walk_steps(steps) if s["kind"] == "valve"]
-    sources = [runner._predicate_source(str(i), s["when"]) for i, s in enumerate(valves)]
+    sources = [WritePredicatesFilter().source(str(i), s["when"]) for i, s in enumerate(valves)]
     name = f"scorecard_predicates_{next(_COUNTER)}"
     path = pathlib.Path(workdir) / f"{name}.py"
     path.write_text("\n\n".join(sources) or "\n")
@@ -71,13 +73,13 @@ def make_predicates(steps, workdir):
 
 # cup: allow-plain the differential test harness: it makes cases, runs them in each mode and compares the results
 def build_pipeline(case, predicates, hooks=()):
-    """The case as a live codeupipe Pipeline (runner.build), with ``hooks``."""
-    pipeline = runner.build(case, Pipeline, predicates, {})
+    """The case as a live codeupipe Pipeline (BuildPipelineFilter), with ``hooks``."""
+    pipeline = BuildPipelineFilter().build(case["steps"], Pipeline, predicates, {})
     for hook in hooks:
         pipeline.use_hook(hook)
     return pipeline
 
 
 def load_export(pipeline, lowering, observe, workdir):
-    """runner._load_export: export, write to disk, import package ``steps``."""
-    return runner._load_export(pipeline, lowering, observe, pathlib.Path(workdir))
+    """ExportFilter: export, write to disk, import package ``steps``."""
+    return ExportFilter().load(pipeline, lowering, observe, pathlib.Path(workdir))
diff --git a/tests/conformance/test_case_runner.py b/tests/conformance/test_case_runner.py
new file mode 100644
index 0000000..414fac4
--- /dev/null
+++ b/tests/conformance/test_case_runner.py
@@ -0,0 +1,110 @@
+"""The Python runner is two pipelines, so each of its steps is tested alone."""
+
+import pathlib
+import sys
+
+import pytest
+
+RUNNER = pathlib.Path(__file__).resolve().parents[2] / "conformance" / "runners" / "python"
+sys.path.insert(0, str(RUNNER))
+
+from codeupipe import Payload, Pipeline  # noqa: E402
+
+import conformance_sink as sink  # noqa: E402
+from case_runner.attach_record_hook_filter import AttachRecordHookFilter  # noqa: E402
+from case_runner.build_pipeline_filter import BuildPipelineFilter  # noqa: E402
+from case_runner.name_steps_filter import NameStepsFilter  # noqa: E402
+from case_runner.parse_target_filter import ParseTargetFilter  # noqa: E402
+from case_runner.run_case_pipeline import RunCasePipeline  # noqa: E402
+from case_runner.run_each_case_filter import RunEachCaseFilter  # noqa: E402
+from case_runner.unsupported import Unsupported  # noqa: E402
+from case_runner.write_predicates_filter import WritePredicatesFilter  # noqa: E402
+
+VALVE = {"kind": "valve", "name": "gate", "when": {"op": "truthy", "key": "on"}, "filter": {"op": "set", "key": "b", "value": 2}}
+CASE = {"id": "t/one", "steps": [{"kind": "filter", "name": "first", "op": "set", "key": "a", "value": 1}, VALVE,
+                                 {"kind": "tap", "name": "audit", "op": "record", "key": "a"}], "input": {"on": True}}
+
+
+@pytest.mark.parametrize("target, compiled, run_async, lowering, observe", [
+    ("run", False, False, "", False), ("run_async", False, True, "", False),
+    ("static-steps", True, False, "steps", False), ("static-inline", True, False, "inline", False),
+    ("static-observe", True, False, "steps", True),
+])
+def test_what_the_name_of_a_target_means(target, compiled, run_async, lowering, observe):
+    out = ParseTargetFilter().call(Payload({"target": target}))
+    assert (out.get("compiled"), out.get("run_async"), out.get("lowering"), out.get("observe")) == (compiled, run_async, lowering, observe)
+
+
+def test_a_name_that_is_not_a_target_stops_the_runner():
+    with pytest.raises(SystemExit, match="unknown target 'nope'"):
+        ParseTargetFilter().call(Payload({"target": "nope"}))
+    with pytest.raises(SystemExit, match="unknown bench target 'run_async'"):
+        ParseTargetFilter(("run",), "bench target").call(Payload({"target": "run_async"}))
+
+
+def built(tmp_path, case=CASE, run_async=False):
+    payload = Payload({"case": case, "workdir": tmp_path, "module_name": f"t_predicates_{abs(hash(tmp_path))}", "run_async": run_async})
+    return BuildPipelineFilter().call(WritePredicatesFilter().call(payload))
+
+
+def test_a_predicate_of_a_case_is_a_real_class_with_test(tmp_path):
+    predicates = WritePredicatesFilter().call(Payload({"case": CASE, "workdir": tmp_path, "module_name": "t_predicates_a"})).get("predicates")
+    assert predicates.when_gate.test(Payload({"on": 1})) is True
+    assert predicates.when_gate.test(Payload({})) is False
+    assert type(predicates.when_gate).__name__ == "When_gate"
+
+
+def test_the_pipeline_of_a_case_and_the_name_of_each_label(tmp_path):
+    out = built(tmp_path)
+    assert isinstance(out.get("pipeline"), Pipeline)
+    assert out.get("label") == {"SetFilter": "first", "SetFilter#2": "gate", "RecordTap": "audit"}
+
+
+@pytest.mark.parametrize("step, message", [
+    ({"kind": "filter", "name": "a", "op": "nope"}, "unknown filter op 'nope'"),
+    ({"kind": "tap", "name": "a", "op": "nope"}, "unknown tap op 'nope'"),
+    ({"kind": "nope", "name": "a"}, "unknown step kind 'nope'"),
+])
+def test_a_word_that_the_runner_does_not_know_is_unsupported(tmp_path, step, message):
+    with pytest.raises(Unsupported, match=message):
+        built(tmp_path, {"id": "t/x", "steps": [step], "input": {}})
+
+
+def test_a_hook_is_attached_for_a_case_that_asks_and_a_target_that_calls_hooks(tmp_path):
+    def hooks(case_hooks, compiled, observe):
+        payload = built(tmp_path, {**CASE, "hooks": case_hooks}).insert("compiled", compiled).insert("observe", observe)
+        return AttachRecordHookFilter().call(payload).get("hooks")
+
+    assert hooks(False, False, False) is None
+    assert len(hooks(True, False, False)) == 1   # the dynamic run
+    assert hooks(True, True, False) is None      # fast compiled code calls no Hooks
+    assert len(hooks(True, True, True)) == 1     # compiled with observe
+
+
+def test_each_step_in_a_result_gets_the_name_of_the_case():
+    result = {"output": {}, "state": {"executed": ["SetFilter", "Other"], "skipped": ["SetFilter#2"], "errors": []},
+              "inner_states": [{"executed": ["SetFilter#2"], "skipped": [], "errors": ["SetFilter"]}],
+              "hooks": ["before:", "before:SetFilter", "error:SetFilter#2"]}
+    named = NameStepsFilter().call(Payload({"result": result, "label": {"SetFilter": "first", "SetFilter#2": "gate"}})).get("result")
+    assert named["state"] == {"executed": ["first", "Other"], "skipped": ["gate"], "errors": []}
+    assert named["inner_states"] == [{"executed": ["gate"], "skipped": [], "errors": ["first"]}]
+    assert named["hooks"] == ["before:", "before:first", "error:gate"]
+
+
+@pytest.mark.parametrize("target", ["run", "run_async", "static-steps", "static-inline", "static-observe"])
+def test_one_case_on_each_target(tmp_path, target):
+    sink.reset()
+    out = RunCasePipeline().run(Payload({"case": CASE, "target": target, "workdir": tmp_path, "module_name": f"t_case_{target.replace('-', '_')}"}))
+    result = out.get("result")
+    assert result["output"] == {"on": True, "a": 1, "b": 2}
+    assert result["taps"] == [1]
+    assert result["input_after"] == {"on": True}
+    assert ("state" in result) == (target not in ("static-steps", "static-inline"))
+    assert result.get("state", {"executed": ["first", "gate", "audit"]})["executed"] == ["first", "gate", "audit"]
+
+
+def test_a_case_that_a_target_cannot_express_is_reported_and_the_next_case_still_runs():
+    cases = [{"id": "t/bad", "steps": [{"kind": "filter", "name": "a", "op": "nope"}], "input": {}}, CASE]
+    results = RunEachCaseFilter(RunCasePipeline()).call(Payload({"cases": cases, "target": "run"})).get("results")
+    assert results[0] == {"id": "t/bad", "unsupported": "unknown filter op 'nope'"}
+    assert results[1]["id"] == "t/one" and results[1]["output"] == {"on": True, "a": 1, "b": 2}
diff --git a/tests/lint/invariants/test_repo_is_clean.py b/tests/lint/invariants/test_repo_is_clean.py
index 0fb44b3..cc7de64 100644
--- a/tests/lint/invariants/test_repo_is_clean.py
+++ b/tests/lint/invariants/test_repo_is_clean.py
@@ -1,13 +1,14 @@
 """This repo's own Python keeps the invariants it lints for."""
 
 import os
+import pathlib
 
 from codeupipe_lint.invariants.lint_pipeline import lint_paths
 
 from ._support import REPO
 
 # The places with a `cup: allow-plain` comment.
-PLAIN_BY_DESIGN = 75
+PLAIN_BY_DESIGN = 60
 
 CHECKED = ["packages/codeupipe/codeupipe", "conformance", "scorecard", "packages/codeupipe-lint/codeupipe_lint", "tests", "examples"]
 
@@ -21,5 +22,9 @@ def test_repo_python_has_no_invariant_violations():
     # the comment gives: a primitive itself, a visitor of a syntax tree, the entry of a program, a test
     # harness. The number is here so that a new one is a change that someone sees.
     allowed = out.get("suppressed")
-    assert {f.rule for f in allowed} == {"CUP-I008"}
-    assert len(allowed) == PLAIN_BY_DESIGN, sorted({f.path for f in allowed})
+    plain = [f for f in allowed if f.rule == "CUP-I008"]
+    assert len(plain) == PLAIN_BY_DESIGN, sorted({f.path for f in plain})
+    # One more comment, for another rule: the conformance runner reads the State of the pipeline of a
+    # case, because that State is what it reports.
+    other = [(f.rule, pathlib.Path(f.path).name) for f in allowed if f.rule != "CUP-I008"]
+    assert other == [("CUP-I003", "run_dynamic_filter.py")]