JoshuaWink · Oct 13, 2026 · Oct 13, 2026 · Oct 13, 2026
diff --git a/README.md b/README.md
@@ -183,7 +183,7 @@ An editable install needs pip 21.3 or later.
 | Command | What it does |
 |---|---|
 | `python -m pytest tests -q` | Runs the tests |
-| `python conformance/run.py` | Runs the 27 conformance cases against the Python dynamic run and the three compiled forms, and against the TypeScript targets that the machine can run |
+| `python conformance/run.py` | Runs the 30 conformance cases against the Python dynamic run and the three compiled forms, and against the TypeScript targets that the machine can run |
 | `python -m codeupipe_lint.anonymous_steps` | Runs the anonymous-step linter. Paths come from `cup.config.json` |
 | `python -m codeupipe_lint.invariants` | Runs the invariant linter. Paths come from `cup.config.json` |
 | `python -m codeupipe.compile examples.pricing.pricing_pipeline:PricingPipeline --out build/pricing` | Compiles the example to `build/pricing/src/steps/` |

diff --git a/conformance/API.md b/conformance/API.md
@@ -120,12 +120,23 @@ A Hook reads the Payload and does not write to it.
 - Order on a filter failure: `… before(step)`, `on_error(step)`,
   `on_error(pipeline)`. No `after` events follow.
 
-## State — what happened in the last run
+## State — what happened in one run
 
 `executed` (labels, in order), `skipped` (labels), `errors`
-(label + error), `timings` (label → seconds, when timing is on). A run
-starts with a new State. **A failed run keeps the steps that completed
-before the failure.**
+(label + error), `timings` (label → seconds, when timing is on). **A failed
+run keeps the steps that completed before the failure.**
+
+**A State belongs to one run, not to the Pipeline.**
+
+- Each run writes into a State of its own. Two runs of one Pipeline that
+  are in progress at the same time never write into the same State.
+- The caller can give the State: `run(p, state)`. The run writes into that
+  State. This is how a caller with more than one run at a time gets the
+  State of each run. The caller gives a new State for each run.
+- With no State from the caller, the run makes a new one.
+- `state` on the Pipeline is the State of the run that started last, with
+  a State from the caller or not. It is for a caller that has one run at a
+  time.
 
 ## Pipeline
 
@@ -137,9 +148,9 @@ before the failure.**
 | add a tap | `add_tap(t)` |
 | nest a pipeline | `add_pipeline(p)` |
 | add a hook | `use_hook(h)` |
-| run (sync) | `run(p)` |
-| run (async, opt-in) | `run_async(p)` |
-| last run | `state` |
+| run (sync) | `run(p)`, `run(p, state)` |
+| run (async, opt-in) | `run_async(p)`, `run_async(p, state)` |
+| the State of the run that started last | `state` |
 | timing and lineage | `observe(timing=True, lineage=False)` |
 | each step's label, in order | `labels` |
 

diff --git a/conformance/MATRIX.md b/conformance/MATRIX.md
@@ -11,7 +11,9 @@ TypeScript run and its compiled form.
 The fast modes (`python:static-steps`, `python:static-inline`,
 `typescript:static`) expose no State or Hooks, so they pass those cases as
 ✅\*. `static-observe` (compile with
-`observe=True`) records both and passes every case outright. See
+`observe=True`) records both and passes every case outright, but for one:
+the case that asks for `state` on the Pipeline object. A compiled form is a
+function and has no such object. See
 [docs/compile.md](../docs/compile.md).
 
 | case | python:run | python:run_async | python:static-steps | python:static-inline | python:static-observe | typescript:run | typescript:static |
@@ -34,6 +36,9 @@ The fast modes (`python:static-steps`, `python:static-inline`,
 | `pipeline/steps-run-in-order` | ✅ | ✅ | ✅* | ✅* | ✅ | ✅ | ✅* |
 | `pipeline/filter-error-halts-the-run` | ✅ | ✅ | ✅* | ✅* | ✅ | ✅ | ✅* |
 | `pipeline/nested-pipeline-is-a-step` | ✅ | ✅ | ✅* | ✅* | ✅ | ✅ | ✅* |
+| `state/the-caller-gives-the-state-of-a-run` | ✅ | ✅ | ✅* | ✅* | ✅ | ✅ | ✅* |
+| `state/a-run-that-starts-inside-another-run-has-its-own-state` | ✅ | ✅ | ✅* | ✅* | ✅ | ✅ | ✅* |
+| `state/with-no-state-from-the-caller-the-pipeline-has-the-state-of-the-run-that-started-last` | ✅ | ✅ | ✅* | ✅* | ✅* | ✅ | ✅* |
 | `tap/observes-without-changing-output` | ✅ | ✅ | ✅* | ✅* | ✅ | ✅ | ✅* |
 | `tap/failure-never-halts-the-pipeline` | ✅ | ✅ | ✅* | ✅* | ✅ | ✅ | ✅* |
 | `tap/failure-does-not-stop-later-taps` | ✅ | ✅ | ✅* | ✅* | ✅ | ✅ | ✅* |
@@ -44,12 +49,12 @@ The fast modes (`python:static-steps`, `python:static-inline`,
 | `valve/sees-upstream-changes` | ✅ | ✅ | ✅* | ✅* | ✅ | ✅ | ✅* |
 | `valve/independent-gates` | ✅ | ✅ | ✅* | ✅* | ✅ | ✅ | ✅* |
 
-- **python:run**: 27 pass, 0 pass*, 0 fail, 0 unsupported
-- **python:run_async**: 27 pass, 0 pass*, 0 fail, 0 unsupported
-- **python:static-steps**: 11 pass, 16 pass*, 0 fail, 0 unsupported
-- **python:static-inline**: 11 pass, 16 pass*, 0 fail, 0 unsupported
-- **python:static-observe**: 27 pass, 0 pass*, 0 fail, 0 unsupported
-- **typescript:run**: 27 pass, 0 pass*, 0 fail, 0 unsupported
-- **typescript:static**: 11 pass, 16 pass*, 0 fail, 0 unsupported
+- **python:run**: 30 pass, 0 pass*, 0 fail, 0 unsupported
+- **python:run_async**: 30 pass, 0 pass*, 0 fail, 0 unsupported
+- **python:static-steps**: 11 pass, 19 pass*, 0 fail, 0 unsupported
+- **python:static-inline**: 11 pass, 19 pass*, 0 fail, 0 unsupported
+- **python:static-observe**: 29 pass, 1 pass*, 0 fail, 0 unsupported
+- **typescript:run**: 30 pass, 0 pass*, 0 fail, 0 unsupported
+- **typescript:static**: 11 pass, 19 pass*, 0 fail, 0 unsupported
 
 ✅* passed everything the target can observe; it does not expose State or Hooks (the fast mode). ➖ the target cannot express the case.
diff --git a/conformance/README.md b/conformance/README.md
@@ -71,6 +71,16 @@ load a number can leave the band.
             "taps": [...], "hooks": ["before:", "before:s1", ...]}}
 ```
 
+Three more keys are for the cases about State and runs that overlap:
+
+- `"given_state": true` in a case: the runner gives the run a new State
+  (`run(p, state)`), and reports that State as `state`.
+- `expect.inner_states`: the State of each run that a `run_again` step
+  started, in order.
+- `expect.pipeline_state`: `state` on the Pipeline after the run ends. A
+  compiled form is a function and has no Pipeline object, so it cannot
+  show this.
+
 Every case also checks, implicitly, that **the caller's input was not mutated**.
 Numbers compare as JSON numbers (`1 == 1.0`), but `true` is never `1`.
 
@@ -87,6 +97,7 @@ Numbers compare as JSON numbers (`1 == 1.0`), but `true` is never `1`.
 | filter | `merge` value | `merge(Payload(value))` |
 | filter | `has` key, to | insert `to = has(key)` |
 | filter | `mutate` key, value | `insert(key, value)` as a statement of its own, then return the payload |
+| filter | `run_again` input | when `get("inner")` is not true: run the pipeline that this step is in one more time, with `input` and a new State, and keep that State for the report. Then return the payload with no change. This is a second run that starts and ends while the first one is in progress. How a pipeline is run again depends on the target (a method of the Pipeline, or for a compiled form a second call of its function), so the runner gives the step that as a function |
 | tap | `record` key | append `get(key)` to the runner's tap log |
 | tap | `fail` message | raise |
 | valve `when` | `equals` key, value | `get(key) == value` |

diff --git a/conformance/cases/state.json b/conformance/cases/state.json
@@ -0,0 +1,202 @@
+[
+ {
+  "id": "state/the-caller-gives-the-state-of-a-run",
+  "given_state": true,
+  "steps": [
+   {
+    "kind": "filter",
+    "name": "first",
+    "op": "set",
+    "key": "a",
+    "value": 1
+   },
+   {
+    "kind": "valve",
+    "name": "gate",
+    "when": {
+     "op": "truthy",
+     "key": "inner"
+    },
+    "filter": {
+     "op": "set",
+     "key": "gated",
+     "value": true
+    }
+   },
+   {
+    "kind": "tap",
+    "name": "audit",
+    "op": "fail",
+    "message": "sink down"
+   },
+   {
+    "kind": "filter",
+    "name": "last",
+    "op": "add",
+    "key": "n",
+    "by": 1
+   }
+  ],
+  "input": {},
+  "expect": {
+   "output": {
+    "a": 1,
+    "n": 1
+   },
+   "state": {
+    "executed": [
+     "first",
+     "audit",
+     "last"
+    ],
+    "skipped": [
+     "gate"
+    ],
+    "errors": [
+     "audit"
+    ]
+   },
+   "inner_states": []
+  }
+ },
+ {
+  "id": "state/a-run-that-starts-inside-another-run-has-its-own-state",
+  "given_state": true,
+  "steps": [
+   {
+    "kind": "filter",
+    "name": "first",
+    "op": "set",
+    "key": "a",
+    "value": 1
+   },
+   {
+    "kind": "filter",
+    "name": "again",
+    "op": "run_again",
+    "input": {
+     "inner": true
+    }
+   },
+   {
+    "kind": "valve",
+    "name": "gate",
+    "when": {
+     "op": "truthy",
+     "key": "inner"
+    },
+    "filter": {
+     "op": "set",
+     "key": "gated",
+     "value": true
+    }
+   },
+   {
+    "kind": "filter",
+    "name": "last",
+    "op": "add",
+    "key": "n",
+    "by": 1
+   }
+  ],
+  "input": {},
+  "expect": {
+   "output": {
+    "a": 1,
+    "n": 1
+   },
+   "state": {
+    "executed": [
+     "first",
+     "again",
+     "last"
+    ],
+    "skipped": [
+     "gate"
+    ],
+    "errors": []
+   },
+   "inner_states": [
+    {
+     "executed": [
+      "first",
+      "again",
+      "gate",
+      "last"
+     ],
+     "skipped": [],
+     "errors": []
+    }
+   ]
+  }
+ },
+ {
+  "id": "state/with-no-state-from-the-caller-the-pipeline-has-the-state-of-the-run-that-started-last",
+  "steps": [
+   {
+    "kind": "filter",
+    "name": "first",
+    "op": "set",
+    "key": "a",
+    "value": 1
+   },
+   {
+    "kind": "filter",
+    "name": "again",
+    "op": "run_again",
+    "input": {
+     "inner": true
+    }
+   },
+   {
+    "kind": "valve",
+    "name": "gate",
+    "when": {
+     "op": "truthy",
+     "key": "inner"
+    },
+    "filter": {
+     "op": "set",
+     "key": "gated",
+     "value": true
+    }
+   },
+   {
+    "kind": "filter",
+    "name": "last",
+    "op": "add",
+    "key": "n",
+    "by": 1
+   }
+  ],
+  "input": {},
+  "expect": {
+   "output": {
+    "a": 1,
+    "n": 1
+   },
+   "inner_states": [
+    {
+     "executed": [
+      "first",
+      "again",
+      "gate",
+      "last"
+     ],
+     "skipped": [],
+     "errors": []
+    }
+   ],
+   "pipeline_state": {
+    "executed": [
+     "first",
+     "again",
+     "gate",
+     "last"
+    ],
+    "skipped": [],
+    "errors": []
+   }
+  }
+ }
+]
diff --git a/conformance/run.py b/conformance/run.py
@@ -164,12 +164,16 @@ def judge(case, result):
             if got != want:
                 problems.append(f"state.{key} {got!r} != {want!r}")
 
+    for key in ("inner_states", "pipeline_state"):
+        if key in expect and key in result and result[key] != expect[key]:
+            problems.append(f"{key} {result[key]!r} != {expect[key]!r}")
+
     if "hooks" in expect and "hooks" in result and result["hooks"] != expect["hooks"]:
         problems.append(f"hooks {result['hooks']!r} != {expect['hooks']!r}")
 
     if problems:
         return "FAIL", problems
-    unchecked = [k for k in ("state", "hooks") if k in expect and k not in result]
+    unchecked = [k for k in ("state", "hooks", "inner_states", "pipeline_state") if k in expect and k not in result]
     if unchecked:
         return "partial", [f"target cannot observe {', '.join(unchecked)}"]
     return "pass", []

diff --git a/conformance/runners/python/conformance_sink.py b/conformance/runners/python/conformance_sink.py
@@ -9,8 +9,14 @@
 
 TAPS = []
 HOOKS = []
+# The State of each run that a run_again step started, in order.
+INNER_STATES = []
+# How the target runs its pipeline one more time: one function, that the runner sets for each case.
+RUN_AGAIN = [None]
 
 
 def reset():
     del TAPS[:]
     del HOOKS[:]
+    del INNER_STATES[:]
+    RUN_AGAIN[0] = None
diff --git a/conformance/runners/python/cup_ops/__init__.py b/conformance/runners/python/cup_ops/__init__.py
@@ -6,6 +6,7 @@
 """
 
 from .add_filter import AddFilter
+from .async_run_again_filter import AsyncRunAgainFilter
 from .append_filter import AppendFilter
 from .copy_filter import CopyFilter
 from .fail_filter import FailFilter
@@ -19,11 +20,13 @@
 from .record_hook import RecordHook
 from .record_tap import RecordTap
 from .remove_filter import RemoveFilter
+from .run_again_filter import RunAgainFilter
 from .set_filter import SetFilter
 
 __all__ = [
     "AddFilter",
     "AppendFilter",
+    "AsyncRunAgainFilter",
     "CopyFilter",
     "FailFilter",
     "FailTap",
@@ -36,5 +39,6 @@
     "RecordHook",
     "RecordTap",
     "RemoveFilter",
+    "RunAgainFilter",
     "SetFilter",
 ]
diff --git a/conformance/runners/python/cup_ops/async_run_again_filter.py b/conformance/runners/python/cup_ops/async_run_again_filter.py
@@ -0,0 +1,15 @@
+"""AsyncRunAgainFilter: the reference `run_again` step for ``run_async`` (conformance/README.md)."""
+
+from conformance_sink import RUN_AGAIN
+
+
+class AsyncRunAgainFilter:
+    """``RunAgainFilter`` for a pipeline that runs with ``run_async``."""
+
+    def __init__(self, data):
+        self.data = data
+
+    async def call(self, payload):
+        if payload.get("inner") is not True:
+            await RUN_AGAIN[0](dict(self.data))
+        return payload