diff --git a/docs/source/core_concepts/flow.md b/docs/source/core_concepts/flow.md
index e9b2f46a..5aa49cbb 100644
--- a/docs/source/core_concepts/flow.md
+++ b/docs/source/core_concepts/flow.md
@@ -86,6 +86,10 @@ MQTT tests can be retried as well, but you should think whether this
 is what you want - you could also try increasing the timeout on an expected MQTT
 response to achieve something similar.
 
+To control _when_ to stop retrying a failing stage, rather than just retrying a fixed number of times, see the
+experimental [`retry_until`](../scripting.md#polling-with-retry_until) and
+[`fail_if`](../scripting.md#failing-fast-with-fail_if) keys.
+
 ## Finalising stages
 
 If you need a stage to run after a test runs, whether it passes or fails (for example, to log out of a service or
diff --git a/docs/source/core_concepts/marks.md b/docs/source/core_concepts/marks.md
index fc2db1f0..c585f9bd 100644
--- a/docs/source/core_concepts/marks.md
+++ b/docs/source/core_concepts/marks.md
@@ -121,6 +121,11 @@ stages:
         n_queries: 10000
 ```
 
+The [`if` key](../scripting.md#running-a-stage-conditionally-with-if) does the same thing, but the logic is inverted
+and it uses Starlark rather than simpleeval. The two cannot both be used on the same stage.
+
+In future, `skip` may be removed in favour of `if`.
+
 ##### Skipping stages with simpleeval expressions
 
 Stages can be skipped by using a `skip` key that contains a [simpleeval](https://pypi.org/project/simpleeval/) expression. 
@@ -140,6 +145,19 @@ stages:
 
 In this example, the stage will be skipped if `v_int` is greater than 50. Any valid simpleeval expression can be used.
 
+The equivalent using the `if` key, which is a Starlark expression rather than a simpleeval one:
+
+```yaml
+stages:
+  - name: Run based on variable value
+    if: "{v_int} <= 50"
+    request:
+      url: "{host}/fake_list"
+      method: GET
+    response:
+      status_code: 200
+```
+
 #### skipif
 
 Sometimes you just want to skip some tests, perhaps based on which server you're
diff --git a/docs/source/scripting.md b/docs/source/scripting.md
index b222d2d0..409a2b9d 100644
--- a/docs/source/scripting.md
+++ b/docs/source/scripting.md
@@ -47,6 +47,14 @@ use.
 To try and combine all of these into one unified test execution model, we need a way to express complex logic
 declaratively, in a format that is more readable than interpolated strings in YAML.
 
+There are two levels to this:
+
+- [Per-stage expressions](#per-stage-expressions) (`if`, `retry_until` and `fail_if`) - keep the normal sequential stage list and
+  just annotate individual stages. This is the closest thing to the GitHub Actions example above and is where you
+  should start.
+- A full [`control_flow` script](#basic-usage) - replaces sequential execution entirely, for things which can't be
+  expressed as a per-stage condition (loops over entities, fallback paths, extracting values with regexes, etc).
+
 ## Starlark Overview
 
 Starlark is a Python-like language designed for configuration and build systems. It provides:
@@ -66,6 +74,235 @@ Starlark control flow is an experimental feature. Enable it with the pytest flag
 pytest --tavern-experimental-starlark-pipeline
 ```
 
+## Per-stage expressions
+
+Rewriting a whole test as a `control_flow` script is a lot of ceremony if all you want is "only run this stage if the
+last one returned something". For that, stages support three keys which are Starlark _expressions_, evaluated by the
+same embedded interpreter. There is no separate script and stages do not need an `id`. These need the same
+`--tavern-experimental-starlark-pipeline` flag as `control_flow`.
+
+Test variables - anything from `save`, `includes`, global config, fixtures, parametrisation, and the `tavern` box - are
+referred to with the same `{format_string}` syntax as everywhere else in Tavern. The expression is interpolated first
+and the result is what gets evaluated, so `if: "{var_x} > 2"` with `var_x` saved as `3` evaluates `3 > 2`. Referring to
+a variable which does not exist is an error, and error messages include both the original expression and the
+interpolated one.
+
+Two things follow from this which are worth being aware of:
+
+> **Quote your strings.** A string variable is interpolated in as-is, not as a Starlark string literal, so write
+> `if: "'{name}' == 'bob'"` rather than `if: "{name} == 'bob'"` - the latter evaluates `bob == 'bob'` and fails with an
+> undefined name.
+>
+> **Escape literal braces.** A Starlark dict or set literal in an expression needs doubled braces, as in
+> `if: "{{'a': 1}}['a'] == 1"`.
+
+Because interpolation happens before evaluation, variable names which are not valid Starlark identifiers - anything
+with a dash in it, or a Starlark reserved word - work fine.
+
+### Multiline expressions
+
+An 'expression' does not have to be one line. Using a YAML block scalar, any of these keys can be a short script, and
+the value of its **last statement** is what decides the result - it still has to be `True` or `False`. Note that only
+an expression has a value, so a script which ends on an assignment (`result = x == 1`) fails rather than using what was
+assigned. Helper modules have to be `load()`ed just like in a `control_flow` script, so this is the way to use `re` in
+a condition:
+
+```yaml
+stages:
+  - name: Only upgrade if the server is on a v2 release
+    if: |
+      load("@tavern_helpers.star", "re")
+
+      match = re.search("v(\\d+)\\.", "{server_banner}")
+      match != None and int(match.groups[0]) == 2
+    request:
+      url: "{global_host}/upgrade"
+      method: POST
+    response:
+      status_code: 200
+```
+
+The same works for `retry_until` and `fail_if`, which additionally have `response` in scope:
+
+```yaml
+    retry_until: |
+      load("@tavern_helpers.star", "re")
+
+      terminal = re.match("(SUCCESS|FAILED)", response.body["status"])
+      terminal != None
+```
+
+> **Be careful with this.** A condition which needs several statements to express is a sign the test is doing quite a
+> lot of thinking, and it is easy to end up with something which is hard to read, hard to debug (Starlark errors are
+> not very helpful, see [Error Messages](#error-messages)), and effectively untested. Prefer a single expression, or a
+> stage which asserts on the response in its `response` block, and only reach for a script when there is no reasonable
+> alternative. If it is getting long, it probably wants to be a [`control_flow` script](#basic-usage) instead.
+
+`run_stage()` is deliberately **not** available - there is already a stage being run - and loading anything other than
+`@tavern_helpers.star` is an error.
+
+### Running a stage conditionally with `if`
+
+The stage only runs if the expression evaluates to `True`. This is an alternative to the
+['skip' key](./core_concepts/marks.md#skipping-stages-with-simpleeval-expressions) - `if` is the same thing with the
+logic inverted, and a stage cannot use both.
+
+```yaml
+stages:
+  - name: Create a user
+    request:
+      url: "{global_host}/users"
+      method: POST
+    response:
+      status_code: 201
+      save:
+        json:
+          n_existing: existing_count
+
+  - name: Only tidy up if there was something there already
+    if: "{n_existing} > 0"
+    request:
+      url: "{global_host}/users/cleanup"
+      method: POST
+    response:
+      status_code: 200
+```
+
+The expression must evaluate to a boolean, and referring to a variable which has not been saved yet is an error rather
+than being treated as false.
+
+`if` is only evaluated for normal stages - stages in a [`finally` block](./core_concepts/flow.md#finalising-stages)
+always run.
+
+### Polling with `retry_until`
+
+`retry_until` is a second opinion on a stage that **failed**. It works like `max_retries`, except that instead of
+blindly retrying it lets you say when to stop:
+
+- If the stage **passes**, it is finished. **`retry_until` is not evaluated at all** - a passing stage is never retried,
+  even if the expression would have been `False`.
+- If the stage **fails**, `retry_until` is evaluated against the response that came back. If it is `True` the stage is
+  treated as finished and the test carries on to the next stage, even though the response block did not match. If it is
+  `False` the stage is retried, up to `max_retries` times, sleeping for `delay_after` in between.
+- If the stage never passes and `retry_until` is never `True`, the test fails.
+
+In other words, adding `retry_until` gives the stage something like the `continue_on_fail` behaviour of
+[`run_stage()`](#run_stage), with the expression deciding when to give up retrying and call it a success.
+
+```yaml
+stages:
+  - name: Poll until the job is ready
+    request:
+      url: "{global_host}/poll"
+      method: GET
+    response:
+      status_code: 200
+      json:
+        status: ready
+    max_retries: 20
+    delay_after: 1
+    retry_until: response.body["status"] == "ready"
+```
+
+As well as the test variables, the expression has a `response` struct in scope with the same properties as the one
+returned by [`run_stage()`](#run_stage):
+
+```starlark
+response.status_code == 200 and response.body["status"] == "{expected_status}"
+```
+
+Because it is an arbitrary expression it can also stop on more than one outcome, which is the usual shape for polling a
+long running job that might end up in any one of several terminal states:
+
+```yaml
+stages:
+  - name: Poll until the job finishes
+    request:
+      url: "{global_host}/job/{job_id}"
+      method: GET
+    response:
+      status_code: 200
+      json:
+        status: SUCCESS
+    max_retries: 20
+    delay_after: 1
+    retry_until: response.body["status"] == "SUCCESS" or response.body["status"] == "FAILED"
+```
+
+Here the `response` block says what the happy path looks like, and `retry_until` says when there is no point polling any
+more. A job which ends up as `FAILED` stops the retries immediately rather than waiting out all 20 of them - but note
+that, as above, a stage which finished because `retry_until` was `True` does not fail the test even though its response
+block did not match. If you need to assert on how the job actually ended, do it in a following stage.
+
+Note that:
+
+- `max_retries` is required - `retry_until` without it is a schema error.
+- Because `retry_until` is only consulted on failure, an expression which is already implied by the `response` block
+  will never be evaluated. Write the `response` block for what you expect once the polling has finished, as in the
+  example above.
+- If the request itself failed and no response was received at all (a connection error, say) there is nothing to
+  evaluate the expression against, so the stage is just retried.
+- Values in the `save` block of an attempt which failed verification are **not** saved, so if a stage finishes because
+  `retry_until` was `True` rather than because it passed, later stages will not see them.
+- `retry_until` does not apply inside a `control_flow` script, which bypasses the retry machinery - use
+  `run_stage(..., continue_on_fail=True)` in a `for` loop instead, as shown in [Retry and Polling](#retry-and-polling).
+
+### Failing fast with `fail_if`
+
+`fail_if` is the mirror image of `retry_until` - a negative assertion which fails the stage as soon as it is `True`:
+
+- It is evaluated after **every** attempt at the stage, whether that attempt passed or failed.
+- If it is `True` the test fails immediately. The stage is **not** retried, no matter what `max_retries` or
+  `retry_until` say.
+- If it is `False` nothing changes - a stage which passed carries on to the next stage, and a stage which failed is
+  retried as normal.
+
+It has the same `response` struct in scope as `retry_until`, which includes `response.failed` if you want to
+distinguish an attempt which passed its response block from one which did not.
+
+The main use for this is polling something which can end up in a state it will never recover from. `retry_until` alone
+can only say "stop polling", which counts as a pass; `fail_if` says "stop polling, and this is a failure":
+
+```yaml
+stages:
+  - name: Poll until the job succeeds
+    request:
+      url: "{global_host}/job/{job_id}"
+      method: GET
+    response:
+      status_code: 200
+      json:
+        status: SUCCESS
+    max_retries: 60
+    delay_after: 10
+    retry_until: response.body["status"] == "SUCCESS"
+    fail_if: response.body["status"] == "FAILED"
+```
+
+A job which goes to `FAILED` fails the test on the next poll instead of spending ten minutes retrying something which
+was never going to succeed.
+
+It is also useful on its own, with no retries at all, as an assertion which is easier to express as an expression than
+as a `response` block:
+
+```yaml
+- name: Check the response does not leak internal errors
+  request:
+    url: "{global_host}/search"
+    method: GET
+  response:
+    status_code: 200
+  fail_if: 'response.body["message"] != None and "traceback" in response.body["message"]'
+```
+
+Note that:
+
+- Unlike `retry_until`, `fail_if` does not need `max_retries`.
+- If the request itself failed and no response was received at all, `fail_if` is not evaluated and the stage fails or
+  retries as it normally would.
+- Like `retry_until`, it does not apply inside a `control_flow` script - check the struct returned by `run_stage()`
+  instead.
+
 ## Basic Usage
 
 ### Inline Control Flow
@@ -388,7 +625,8 @@ control_flow: |
 **Important:** Starlark control flow currently only works with HTTP/REST tests. Other protocol backends (MQTT, gRPC,
 GraphQL) are not yet supported.
 
-Attempting to use `run_stage()` with non-HTTP stages will raise a `NotImplementedError`.
+Attempting to use `run_stage()` - or `retry_until`/`fail_if` - with non-HTTP stages will raise a `NotImplementedError`.
+The `if` key works with any backend, as it only sees test variables.
 
 ### Error Messages
 
@@ -433,7 +671,7 @@ Key differences from Python:
 ## Examples
 
 See the integration test files in `tests/integration/starlark/` for complete working examples of basic control flow,
-includes, regex extraction, retry patterns
+includes, regex extraction, retry patterns, and the per-stage `if`/`retry_until` keys.
 
 ## Possible future improvements
 
@@ -446,3 +684,4 @@ includes, regex extraction, retry patterns
     - Make this auto-export functions into either this document with mkdocstrings into
     - Let users import their own functions into starlark?
 - Add a new CLI/ini flag to say "run 'finally' stages when using starlark script"
+- Allow `if` on `finally` stages, and give it access to the previous stage's response.
diff --git a/tavern/_core/exceptions.py b/tavern/_core/exceptions.py
index 2c99f6a7..e7e1ab81 100644
--- a/tavern/_core/exceptions.py
+++ b/tavern/_core/exceptions.py
@@ -1,4 +1,4 @@
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING, Any, Optional
 
 if TYPE_CHECKING:
     from tavern._core.pytest.config import TestConfig
@@ -13,11 +13,13 @@ class TavernException(Exception):
         is_final: whether this exception came from a 'finally' block
         stage: stage that caused this issue
         test_block_config: config for stage
+        response: the response from the stage, if one was received before the failure
     """
 
     stage: Optional[dict]
     test_block_config: Optional["TestConfig"]
     is_final: bool = False
+    response: Optional[Any] = None
 
 
 class BadSchemaError(TavernException):
@@ -36,6 +38,13 @@ def __init__(self, msg, failures=None) -> None:
         self.failures = failures or []
 
 
+class FailIfError(TestFailError):
+    """A stage's 'fail_if' expression was true
+
+    This is separate from a normal test failure because it should never be retried
+    """
+
+
 class KeyMismatchError(TavernException):
     """Mismatch found while validating keys in response"""
 
diff --git a/tavern/_core/run.py b/tavern/_core/run.py
index 4fe57db8..5899e11a 100644
--- a/tavern/_core/run.py
+++ b/tavern/_core/run.py
@@ -62,7 +62,7 @@ def _run_with_starlark_control_flow(
         import starlark
     except ImportError as e:
         raise exceptions.DependencyMissingError(
-            "starlark", "pip install tavern[starlark]"
+            "starlark", "pip install tavern[scriptable]"
         ) from e
 
     from tavern._core.starlark.starlark_env import StarlarkPipelineRunner
@@ -326,6 +326,14 @@ def getonly(stage):
                     if eval_skip(content, test_block_config):
                         continue
 
+                if (condition := stage.get("if")) is not None:
+                    if not _eval_stage_condition(condition, stage, test_block_config):
+                        logger.info(
+                            "Skipping stage '%s' as 'if' condition was false",
+                            stage["name"],
+                        )
+                        continue
+
                 if has_only and not getonly(stage):
                     continue
 
@@ -352,6 +360,73 @@ def getonly(stage):
                 logger.debug("no 'finally' stages to run")
 
 
+def _eval_stage_condition(
+    condition: str, stage: Mapping, test_block_config: TestConfig
+) -> bool:
+    """Evaluate the 'if' key on a stage to see whether it should be run
+
+    Args:
+        condition: Starlark expression from the 'if' key
+        stage: the stage it came from
+        test_block_config: current test config
+
+    Returns:
+        Whether the stage should be run
+    """
+    # Local import to avoid a circular dependency, and to keep starlark optional
+    from tavern._core.starlark.expressions import eval_stage_expression
+
+    if not isinstance(condition, str):
+        raise exceptions.BadSchemaError(
+            f"Unexpected '{type(condition)}' in if key - should be a string"
+        )
+
+    return eval_stage_expression("if", condition, stage, test_block_config)
+
+
+def _check_fail_if(
+    fail_if: str,
+    stage: Mapping,
+    test_block_config: TestConfig,
+    response: Any,
+    *,
+    success: bool,
+) -> None:
+    """Evaluate the 'fail_if' key on a stage against the response it got back
+
+    Args:
+        fail_if: Starlark expression from the 'fail_if' key
+        stage: the stage it came from
+        test_block_config: current test config
+        response: the response from running the stage
+        success: whether the stage passed all of its verifications
+
+    Raises:
+        exceptions.FailIfError: if the expression was true
+    """
+    # Local import to avoid a circular dependency, and to keep starlark optional
+    from tavern._core.starlark.expressions import eval_response_expression
+
+    if not eval_response_expression(
+        "fail_if",
+        fail_if,
+        stage,
+        test_block_config,
+        response=response,
+        success=success,
+        request_vars=test_block_config.variables,
+    ):
+        return
+
+    error = exceptions.FailIfError(
+        "Test '{}' failed: 'fail_if' expression was true: {}".format(
+            stage["name"], fail_if
+        )
+    )
+    error.response = response
+    raise error
+
+
 def _calculate_stage_strictness(
     stage: dict, test_block_config: TestConfig, test_spec: Mapping
 ) -> StrictLevel:
@@ -508,11 +583,27 @@ def wrapped_run_stage(
 
         tinctures.end_tinctures(expected, response)
 
-        for response_type, response_verifiers in verifiers.items():
-            logger.debug("Running verifiers for %s", response_type)
-            for v in response_verifiers:
-                saved = v.verify(response)
-                stage_config.variables.update(saved)
+        fail_if = stage.get("fail_if", None)
+
+        try:
+            for response_type, response_verifiers in verifiers.items():
+                logger.debug("Running verifiers for %s", response_type)
+                for v in response_verifiers:
+                    saved = v.verify(response)
+                    stage_config.variables.update(saved)
+        except exceptions.TavernException as e:
+            # Attach the response so that things like 'retry_until' can still inspect it
+            # even though the stage failed verification
+            e.response = response
+
+            # A stage which failed can still be in a state which should not be retried
+            if fail_if is not None:
+                _check_fail_if(fail_if, stage, stage_config, response, success=False)
+
+            raise
+
+        if fail_if is not None:
+            _check_fail_if(fail_if, stage, stage_config, response, success=True)
 
         tavern_box.pop("request_vars")
         delay(stage, "after", stage_config.variables)
diff --git a/tavern/_core/schema/tests.jsonschema.yaml b/tavern/_core/schema/tests.jsonschema.yaml
index 3534ff7e..98f48787 100644
--- a/tavern/_core/schema/tests.jsonschema.yaml
+++ b/tavern/_core/schema/tests.jsonschema.yaml
@@ -94,6 +94,17 @@ definitions:
     required:
       - name
 
+    # 'retry_until' is meaningless without something to bound the number of retries
+    dependencies:
+      retry_until:
+        - max_retries
+
+    # 'if' is the starlark equivalent of the older simpleeval 'skip' key
+    not:
+      required:
+        - skip
+        - if
+
     properties:
       tinctures:
         type: array
@@ -117,7 +128,24 @@ definitions:
             default: false
 
           - type: string
-            description: CEL expression saying whether to skip this stage
+            description: simpleeval expression saying whether to skip this stage
+
+      if:
+        type: string
+        description: Starlark expression - this stage is only run if it evaluates to True
+
+      retry_until:
+        type: string
+        description:
+          Starlark expression evaluated after each *failed* attempt at this stage - the
+          stage is retried until it evaluates to True, up to max_retries times. Not
+          evaluated if the stage passes.
+
+      fail_if:
+        type: string
+        description:
+          Starlark expression evaluated after every attempt at this stage - if it
+          evaluates to True the test fails immediately, without any further retries.
 
       only:
         type: boolean
diff --git a/tavern/_core/starlark/builtins.py b/tavern/_core/starlark/builtins.py
new file mode 100644
index 00000000..662749db
--- /dev/null
+++ b/tavern/_core/starlark/builtins.py
@@ -0,0 +1,125 @@
+"""Bindings for the helper 'library' modules loaded from tavern_helpers.star.
+
+These are the parts of the Starlark environment which do not need a pipeline runner -
+'re', 'time' and 'log'. They are shared between a full 'control_flow' script and the
+per-stage expressions, which can also load them but cannot run stages.
+
+This module must not import anything from tavern._core.run, so that it can be imported
+from the per-stage expression path.
+"""
+
+import functools
+import importlib.resources
+import logging
+import re
+import time
+from typing import TYPE_CHECKING, Any
+
+from .types import from_starlark, to_starlark
+
+if TYPE_CHECKING:
+    import starlark
+
+logger: logging.Logger = logging.getLogger(__name__)
+
+
+def wrap_callable(fn):
+    """Decorator that converts all arguments from starlark→Python before
+    calling *fn*, and converts the return value from Python→starlark."""
+
+    @functools.wraps(fn)
+    def wrapper(*args, **kwargs):
+        converted_args = [from_starlark(a) for a in args]
+        converted_kwargs = {k: from_starlark(v) for k, v in kwargs.items()}
+        result = fn(*converted_args, **converted_kwargs)
+        return to_starlark(result)
+
+    return wrapper
+
+
+def get_starlark_builtins() -> str:
+    """Load the Starlark builtins from the tavern_helpers.star file.
+
+    Returns:
+        The Starlark code for built-in helper functions
+    """
+    return (
+        importlib.resources.files(__package__)
+        .joinpath("tavern_helpers.star")
+        .read_text()
+    )
+
+
+def _match_to_dict(result: "re.Match | None") -> dict | None:
+    if result is None:
+        return None
+    return {
+        "group0": result.group(0),
+        "groups": list(result.groups()),
+        "start": result.start(),
+        "end": result.end(),
+    }
+
+
+def add_library_callables(module: "starlark.Module") -> None:
+    """Add the dunder bindings which the 're', 'time' and 'log' helpers wrap
+
+    Args:
+        module: the starlark module to add them to
+    """
+
+    @wrap_callable
+    def log(s: str) -> None:
+        """log a string to stdout."""
+        logger.info(s)
+
+    module.add_callable("log", log)
+
+    @wrap_callable
+    def re_match(pattern: str, string: str | bytes) -> dict | None:
+        if isinstance(string, bytes):
+            string = string.decode("utf-8")
+        return _match_to_dict(re.match(pattern, string))
+
+    module.add_callable("__re_match", re_match)
+
+    @wrap_callable
+    def re_search(pattern: str, string: str | bytes) -> dict | None:
+        if isinstance(string, bytes):
+            string = string.decode("utf-8")
+        return _match_to_dict(re.search(pattern, string))
+
+    module.add_callable("__re_search", re_search)
+
+    @wrap_callable
+    def re_sub(pattern: str, repl: str, string: str | bytes) -> str:
+        if isinstance(string, bytes):
+            return re.sub(pattern, repl, string.decode("utf-8"))
+        return re.sub(pattern, repl, string)
+
+    module.add_callable("__re_sub", re_sub)
+
+    @wrap_callable
+    def time_sleep(seconds: float) -> None:
+        time.sleep(seconds)
+
+    module.add_callable("__time_sleep", time_sleep)
+
+
+def add_unavailable_run_stage(module: "starlark.Module", reason: str) -> None:
+    """Bind a 'run_stage' which just explains why it can't be used
+
+    tavern_helpers.star always defines 'run_stage', so somewhere which can't run stages
+    still has to bind something for it to call.
+
+    Args:
+        module: the starlark module to add it to
+        reason: message explaining why running a stage isn't possible here
+    """
+    from tavern._core import exceptions
+
+    @wrap_callable
+    def run_stage_unavailable(*args: Any, **kwargs: Any) -> Any:
+        raise exceptions.StarlarkError(reason)
+
+    module.add_callable("__run_stage", run_stage_unavailable)
diff --git a/tavern/_core/starlark/expressions.py b/tavern/_core/starlark/expressions.py
new file mode 100644
index 00000000..626fcd30
--- /dev/null
+++ b/tavern/_core/starlark/expressions.py
@@ -0,0 +1,282 @@
+"""Evaluation of single Starlark expressions embedded in a stage.
+
+This is used for the per-stage ``if``, ``retry_until`` and ``fail_if`` keys, which are a
+much lighter-weight alternative to writing a whole ``control_flow`` script. Like the
+simpleeval based ``skip`` key, expressions here are format-string interpolated before
+being evaluated, so ``if: "{var_x} > 2"`` is the way to refer to a test variable.
+
+An 'expression' can also be several statements long, in which case the value of the last
+one is what decides the result. The helper modules can be loaded as in a ``control_flow``
+script, but ``run_stage`` is not available - there is already a stage being run.
+
+This module must not import anything from tavern._core.run, and must not import
+starlark at the top level, so that it can be imported (lazily) from the normal
+non-starlark test path.
+"""
+
+import logging
+from collections.abc import Mapping
+from typing import TYPE_CHECKING, Any
+
+from tavern._core import exceptions
+from tavern._core.dict_util import format_keys
+
+if TYPE_CHECKING:
+    import starlark
+
+    from tavern._core.pytest.config import TestConfig
+
+logger: logging.Logger = logging.getLogger(__name__)
+
+# Name the response dict is bound to before being turned into a struct
+_RESPONSE_DICT_NAME = "__tavern_response"
+
+_RESPONSE_PRELUDE = f"response = struct(**{_RESPONSE_DICT_NAME})"
+
+
+def _import_starlark():
+    """Import the starlark module, raising a useful error if it isn't installed"""
+    try:
+        import starlark
+    except ImportError as e:
+        raise exceptions.DependencyMissingError(
+            "starlark", "pip install tavern[scriptable]"
+        ) from e
+
+    return starlark
+
+
+def _get_dialect() -> "starlark.Dialect":
+    starlark = _import_starlark()
+    dialect = starlark.Dialect.extended()
+    dialect.enable_keyword_only_arguments = True
+    return dialect
+
+
+def _get_globals() -> "starlark.Globals":
+    starlark = _import_starlark()
+    return starlark.Globals.standard().extended_by(
+        [
+            starlark.LibraryExtension.StructType,
+        ]
+    )
+
+
+def _get_file_loader(
+    module_globals: "starlark.Globals", dialect: "starlark.Dialect"
+) -> "starlark.FileLoader":
+    """Get a loader which makes the tavern helper modules available to an expression
+
+    This is the same set of helpers as in a 'control_flow' script, except that
+    'run_stage' can't do anything - the stage the expression is attached to is already
+    being run.
+
+    Args:
+        module_globals: globals to evaluate the helpers with
+        dialect: dialect to parse the helpers with
+
+    Returns:
+        a loader which handles '@tavern_helpers.star'
+    """
+    starlark = _import_starlark()
+
+    from .builtins import (
+        add_library_callables,
+        add_unavailable_run_stage,
+        get_starlark_builtins,
+    )
+
+    # The return type is a starlark.FrozenModule, but the name is shadowed by the
+    # local import above
+    def load(filename: str) -> Any:
+        if filename != "@tavern_helpers.star":
+            raise FileNotFoundError(filename)
+
+        helpers = starlark.Module()
+        add_library_callables(helpers)
+        add_unavailable_run_stage(
+            helpers,
+            "'run_stage' is not available in a per-stage expression - use a "
+            "'control_flow' script if you need to run another stage",
+        )
+        ast = starlark.parse(filename, get_starlark_builtins(), dialect=dialect)
+        starlark.eval(helpers, ast, module_globals)
+
+        return helpers.freeze()
+
+    return starlark.FileLoader(load)
+
+
+def eval_stage_expression(
+    key: str,
+    expr: str,
+    stage: Mapping[str, Any],
+    test_block_config: "TestConfig",
+    *,
+    response: Mapping[str, Any] | None = None,
+) -> bool:
+    """Evaluate the Starlark expression from a per-stage 'if' or 'retry_until' key
+
+    Args:
+        key: name of the stage key the expression came from
+        expr: the Starlark expression
+        stage: the stage it came from, used for error messages
+        test_block_config: current test config, the variables from which are bound as
+            globals in the expression
+        response: response values to bind as a 'response' struct, if any
+
+    Returns:
+        the result of the expression
+
+    Raises:
+        exceptions.UnexpectedKeysError: if the experimental starlark pipeline was not enabled
+        exceptions.EvalError: if the expression could not be run, or did not evaluate to
+            a boolean
+    """
+    if not test_block_config.experimental_starlark_pipeline:
+        raise exceptions.UnexpectedKeysError(
+            f"'{key}' requires the experimental starlark pipeline to be enabled - pass "
+            "--tavern-experimental-starlark-pipeline or set "
+            "tavern-experimental-starlark-pipeline in your pytest ini file"
+        )
+
+    return eval_expression(
+        expr,
+        test_block_config.variables,
+        response=response,
+        description=f"'{key}' in stage '{stage.get('name', 'unnamed-stage')}'",
+    )
+
+
+def eval_response_expression(
+    key: str,
+    expr: str,
+    stage: Mapping[str, Any],
+    test_block_config: "TestConfig",
+    *,
+    response: Any,
+    success: bool,
+    request_vars: Mapping[str, Any],
+) -> bool:
+    """Evaluate a stage expression which can also inspect the response from the stage
+
+    This is used for the 'retry_until' and 'fail_if' keys, which both get a 'response'
+    struct bound in the expression.
+
+    Args:
+        key: name of the stage key the expression came from
+        expr: the Starlark expression
+        stage: the stage it came from
+        test_block_config: current test config
+        response: the response from running the stage, if any
+        success: whether the stage passed all of its verifications
+        request_vars: any variables captured during the request
+
+    Returns:
+        the result of the expression
+
+    Raises:
+        exceptions.BadSchemaError: if the expression was not a string
+    """
+    from .response_struct import create_response_struct
+
+    if not isinstance(expr, str):
+        raise exceptions.BadSchemaError(
+            f"Unexpected '{type(expr)}' in {key} key - should be a string"
+        )
+
+    response_values = create_response_struct(
+        response,
+        success=success,
+        request_vars=dict(request_vars),
+        stage_name=stage.get("name", "unnamed-stage"),
+    )
+
+    return eval_stage_expression(
+        key,
+        expr,
+        stage,
+        test_block_config,
+        response=response_values,
+    )
+
+
+def eval_expression(
+    expr: str,
+    variables: Mapping[str, Any],
+    *,
+    response: Mapping[str, Any] | None = None,
+    description: str,
+) -> bool:
+    """Evaluate a Starlark expression, interpolating the given variables into it first
+
+    Args:
+        expr: the Starlark expression to evaluate, which may contain format strings
+            referring to test variables
+        variables: test variables to interpolate into the expression
+        response: if given, a dict of response values (see
+            :func:`tavern._core.starlark.response_struct.create_response_struct`) which
+            is bound as a struct called 'response'
+        description: what this expression is, used in error messages
+
+    Returns:
+        the result of the expression
+
+    Raises:
+        exceptions.EvalError: if the expression could not be formatted or run, or if it
+            did not evaluate to a boolean
+    """
+    starlark = _import_starlark()
+
+    from .types import from_starlark, to_starlark
+
+    try:
+        formatted = format_keys(expr, variables)
+    except exceptions.MissingFormatError as e:
+        raise exceptions.EvalError(
+            f"Undefined variable used in Starlark expression for {description}: {expr}"
+        ) from e
+
+    dialect = _get_dialect()
+    module = starlark.Module()
+    module_globals = _get_globals()
+
+    if response is not None:
+        module[_RESPONSE_DICT_NAME] = to_starlark(dict(response))
+        prelude = starlark.parse(description, _RESPONSE_PRELUDE, dialect=dialect)
+        starlark.eval(module, prelude, module_globals)
+
+    try:
+        ast = starlark.parse(description, formatted, dialect=dialect)
+    except starlark.StarlarkError as e:
+        raise exceptions.EvalError(
+            f"Error parsing Starlark expression for {description}: {formatted} "
+            f"(from {expr})"
+        ) from e
+
+    logger.debug(
+        "Evaluating Starlark expression for %s: %s (from %s)",
+        description,
+        formatted,
+        expr,
+    )
+
+    try:
+        result = starlark.eval(
+            module, ast, module_globals, _get_file_loader(module_globals, dialect)
+        )
+    except starlark.StarlarkError as e:
+        raise exceptions.EvalError(
+            f"Error evaluating Starlark expression for {description}: {formatted} "
+            f"(from {expr}) ({e})"
+        ) from e
+
+    result = from_starlark(result)
+
+    if not isinstance(result, bool):
+        raise exceptions.EvalError(
+            f"Starlark expression for {description} did not evaluate to True/False "
+            f"(got {result} of type {type(result)}): {formatted} (from {expr})"
+        )
+
+    return result
diff --git a/tavern/_core/starlark/response_struct.py b/tavern/_core/starlark/response_struct.py
new file mode 100644
index 00000000..ff3ae6e2
--- /dev/null
+++ b/tavern/_core/starlark/response_struct.py
@@ -0,0 +1,69 @@
+"""Conversion of a plugin response object into a dict for use in Starlark.
+
+This is used both by the full ``control_flow`` pipeline (where it becomes the struct
+returned by ``run_stage()``) and by per-stage ``retry_until`` expressions.
+"""
+
+from typing import Any
+
+import requests
+
+
+def create_response_struct(
+    response: Any | None,
+    *,
+    success: bool,
+    request_vars: dict[str, Any],
+    stage_name: str,
+) -> dict[str, Any]:
+    """Convert a response from running a stage into a dict of Starlark-safe values.
+
+    The returned dict is intended to be splatted into a Starlark ``struct()`` so that
+    users can write ``response.status_code`` etc.
+
+    Args:
+        response: the response from the plugin that ran the stage, if any
+        success: whether the stage passed all of its verifications
+        request_vars: any variables captured during the request
+        stage_name: name of the stage that was run
+
+    Returns:
+        dict of response values
+
+    Raises:
+        NotImplementedError: if the response is from a plugin which is not supported yet
+    """
+    base_dict: dict[str, Any] = {
+        # Add "failed" so people don't have to do "if not resp.success" when people will almost certainly
+        # want to do "if resp.failed" most of the time
+        "failed": not success,
+        "success": success,
+        "request_vars": request_vars,
+        "stage_name": stage_name,
+    }
+
+    if response is None:
+        return base_dict
+
+    if isinstance(response, requests.Response):
+        content_type = response.headers.get("Content-Type", "")
+
+        # Try to parse JSON body, fall back to raw content
+        if "application/json" in content_type:
+            body = response.json()
+        else:
+            body = response.content
+
+        base_dict.update(
+            {
+                "status_code": response.status_code,
+                "body": body,
+                "headers": response.headers,
+                "cookies": response.cookies,
+            }
+        )
+        return base_dict
+
+    raise NotImplementedError(
+        f"gRPC, MQTT, etc. are not supported yet. Got {type(response)}"
+    )
diff --git a/tavern/_core/starlark/starlark_env.py b/tavern/_core/starlark/starlark_env.py
index 6e4e052f..56cfa3b9 100644
--- a/tavern/_core/starlark/starlark_env.py
+++ b/tavern/_core/starlark/starlark_env.py
@@ -2,14 +2,9 @@
 
 import copy
 import dataclasses
-import functools
-import importlib.resources
 import logging
-import re
-import time
 from typing import Any, TypedDict
 
-import requests
 import starlark
 
 from tavern._core import exceptions
@@ -19,26 +14,14 @@
 from tavern._core.strict_util import StrictLevel
 from tavern._core.tincture import get_stage_tinctures
 
+from .builtins import add_library_callables, get_starlark_builtins, wrap_callable
+from .response_struct import create_response_struct
 from .stage_registry import StageRegistry
 from .types import from_starlark, to_starlark
 
 logger: logging.Logger = logging.getLogger(__name__)
 
 
-def _wrap_callable(fn):
-    """Decorator that converts all arguments from starlark→Python before
-    calling *fn*, and converts the return value from Python→starlark."""
-
-    @functools.wraps(fn)
-    def wrapper(*args, **kwargs):
-        converted_args = [from_starlark(a) for a in args]
-        converted_kwargs = {k: from_starlark(v) for k, v in kwargs.items()}
-        result = fn(*converted_args, **converted_kwargs)
-        return to_starlark(result)
-
-    return wrapper
-
-
 class PipelineContext(TypedDict):
     """Context object passed between stages in starlark pipelines.
 
@@ -88,19 +71,6 @@ def from_starlark(cls, obj: dict) -> "StageResponse":
         )
 
 
-def _get_starlark_builtins() -> str:
-    """Load the Starlark builtins from the tavern_helpers.star file.
-
-    Returns:
-        The Starlark code for built-in helper functions
-    """
-    return (
-        importlib.resources.files(__package__)
-        .joinpath("tavern_helpers.star")
-        .read_text()
-    )
-
-
 class StarlarkPipelineRunner:
     """Runner for executing starlark pipeline scripts.
 
@@ -163,9 +133,7 @@ def load_and_run(self, script: str) -> Any:
         def load(filename: str) -> starlark.FrozenModule:
             """Implements the 'load' function in starlark. Currently only supports loading tavern helpers."""
             if filename == "@tavern_helpers.star":
-                ast = starlark.parse(
-                    filename, _get_starlark_builtins(), dialect=dialect
-                )
+                ast = starlark.parse(filename, get_starlark_builtins(), dialect=dialect)
                 mod = starlark.Module()
                 self._setup_builtins(mod)
                 starlark.eval(mod, ast, self.globals)
@@ -254,38 +222,11 @@ def _run_stage(
 
     def _create_response_struct(self, stage_response: StageResponse) -> dict[str, Any]:
         """Convert StageResponse to dict that starlark converts to struct."""
-        base_dict: dict[str, Any] = {
-            # Add "failed" so people don't have to do "if not resp.success" when people will almost certainly
-            # want to do "if resp.failed" most of the time
-            "failed": not stage_response.success,
-            "success": stage_response.success,
-            "request_vars": stage_response.request_vars,
-            "stage_name": stage_response.stage_name,
-        }
-        if stage_response.response is None:
-            return base_dict
-        elif isinstance(stage_response.response, requests.Response):
-            rest_response = stage_response.response
-            content_type = rest_response.headers.get("Content-Type", "")
-
-            # Try to parse JSON body, fall back to raw content
-            if "application/json" in content_type:
-                body = rest_response.json()
-            else:
-                body = rest_response.content
-
-            base_dict.update(
-                {
-                    "status_code": rest_response.status_code,
-                    "body": body,
-                    "headers": rest_response.headers,
-                    "cookies": rest_response.cookies,
-                }
-            )
-            return base_dict
-
-        raise NotImplementedError(
-            f"gRPC, MQTT, etc. are not supported yet. Got {type(stage_response.response)}"
+        return create_response_struct(
+            stage_response.response,
+            success=stage_response.success,
+            request_vars=stage_response.request_vars,
+            stage_name=stage_response.stage_name,
         )
 
     def _setup_builtins(self, module: "starlark.Module") -> None:
@@ -316,14 +257,16 @@ def _re_sub(pattern, repl, s):
 
             re = struct(match=_re_match, sub=_re_sub)
 
-        2. Add a wrapper function into this function and add it with module.add_callable.
-           dunder names are used to 'hide' the original function from the user.
+        2. Add a wrapper function into builtins.add_library_callables (or into this
+           function, if it needs the pipeline runner) and add it with
+           module.add_callable. dunder names are used to 'hide' the original function
+           from the user.
 
-            @_wrap_callable
+            @wrap_callable
             def re_match(pattern, s):
                 return re.match(pattern, s)
 
-            @_wrap_callable
+            @wrap_callable
             def re_sub(pattern, repl, s):
                 return re.sub(pattern, repl, s)
 
@@ -339,10 +282,12 @@ def re_sub(pattern, repl, s):
             if not re.match("(one_thing|another_thing)", resp.json["key"]):
                 fail("No match found")
         """
+        add_library_callables(module)
+
         for stage_id, stage in self._stage_registry.get_all_stages().items():
             module[stage_id] = to_starlark(stage)
 
-        @_wrap_callable
+        @wrap_callable
         def run_stage_binding(
             stage_id: str, continue_on_fail: bool, extra_vars: dict | None
         ) -> Any:
@@ -364,56 +309,3 @@ def run_stage_binding(
                 ) from e
 
         module.add_callable("__run_stage", run_stage_binding)
-
-        @_wrap_callable
-        def log(s: str) -> None:
-            """log a string to stdout."""
-            logger.info(s)
-
-        module.add_callable("log", log)
-
-        @_wrap_callable
-        def re_match(pattern: str, string: str | bytes) -> dict | None:
-            if isinstance(string, bytes):
-                string = string.decode("utf-8")
-            result = re.match(pattern, string)
-            if result is None:
-                return None
-            return {
-                "group0": result.group(0),
-                "groups": list(result.groups()),
-                "start": result.start(),
-                "end": result.end(),
-            }
-
-        module.add_callable("__re_match", re_match)
-
-        @_wrap_callable
-        def re_search(pattern: str, string: str | bytes) -> dict | None:
-            if isinstance(string, bytes):
-                string = string.decode("utf-8")
-            result = re.search(pattern, string)
-            if result is None:
-                return None
-            return {
-                "group0": result.group(0),
-                "groups": list(result.groups()),
-                "start": result.start(),
-                "end": result.end(),
-            }
-
-        module.add_callable("__re_search", re_search)
-
-        @_wrap_callable
-        def re_sub(pattern: str, repl: str, string: str | bytes) -> str:
-            if isinstance(string, bytes):
-                return re.sub(pattern, repl, string.decode("utf-8"))
-            return re.sub(pattern, repl, string)
-
-        module.add_callable("__re_sub", re_sub)
-
-        @_wrap_callable
-        def time_sleep(seconds: float) -> None:
-            time.sleep(seconds)
-
-        module.add_callable("__time_sleep", time_sleep)
diff --git a/tavern/_core/testhelpers.py b/tavern/_core/testhelpers.py
index 93cfa4a7..109880af 100644
--- a/tavern/_core/testhelpers.py
+++ b/tavern/_core/testhelpers.py
@@ -2,6 +2,7 @@
 import time
 from collections.abc import Callable, Mapping
 from functools import wraps
+from typing import Any
 
 from tavern._core import exceptions
 from tavern._core.dict_util import format_keys
@@ -28,6 +29,37 @@ def delay(stage: Mapping, when: str, variables: Mapping) -> None:
         time.sleep(length)
 
 
+def _check_retry_until(
+    retry_until: str,
+    stage: Mapping,
+    test_block_config: TestConfig,
+    response: Any,
+) -> bool:
+    """Evaluate the 'retry_until' expression against the response from a failed stage
+
+    Args:
+        retry_until: Starlark expression from the 'retry_until' key
+        stage: test stage
+        test_block_config: Configuration for current test
+        response: the response from the attempt that just failed
+
+    Returns:
+        Whether the stage should be considered finished anyway
+    """
+    # Local import to avoid a circular dependency, and to keep starlark optional
+    from tavern._core.starlark.expressions import eval_response_expression
+
+    return eval_response_expression(
+        "retry_until",
+        retry_until,
+        stage,
+        test_block_config,
+        response=response,
+        success=False,
+        request_vars=test_block_config.variables,
+    )
+
+
 def retry(stage: Mapping, test_block_config: TestConfig) -> Callable:
     """Look for retry and try to repeat the stage `retry` times.
 
@@ -41,6 +73,14 @@ def retry(stage: Mapping, test_block_config: TestConfig) -> Callable:
     else:
         max_retries = 0
 
+    retry_until = stage.get("retry_until", None)
+
+    if retry_until and max_retries == 0:
+        raise exceptions.InvalidRetryException(
+            f"Stage '{stage['name']}' used 'retry_until' but max_retries was 0 - "
+            "'retry_until' requires a nonzero 'max_retries'"
+        )
+
     if max_retries == 0:
 
         def catch_wrapper(fn):
@@ -65,7 +105,33 @@ def wrapped(*args, **kwargs):
                         res = fn(*args, **kwargs)
                     except exceptions.BadSchemaError:
                         raise
+                    except exceptions.FailIfError:
+                        # 'fail_if' is a terminal state, there is no point retrying
+                        logger.error(
+                            "Stage '%s' matched its 'fail_if' expression, not retrying.",
+                            stage["name"],
+                        )
+                        raise
                     except exceptions.TavernException as e:
+                        # The stage failed, so if there's a 'retry_until' expression see
+                        # whether it considers the stage finished anyway
+                        if retry_until:
+                            if e.response is not None:
+                                if _check_retry_until(
+                                    retry_until, stage, test_block_config, e.response
+                                ):
+                                    logger.info(
+                                        "Stage '%s' failed but 'retry_until' was true, continuing.",
+                                        stage["name"],
+                                    )
+                                    res = e.response
+                                    break
+                            else:
+                                logger.debug(
+                                    "No response from stage '%s' so 'retry_until' could not be evaluated",
+                                    stage["name"],
+                                )
+
                         if i < max_retries:
                             logger.info(
                                 "Stage '%s' failed for %i time. Retrying.",
@@ -80,7 +146,13 @@ def wrapped(*args, **kwargs):
                                 max_retries,
                             )
 
-                            if isinstance(e, exceptions.TestFailError):
+                            if retry_until:
+                                raise exceptions.TestFailError(
+                                    "Test '{}' failed: stage did not succeed and 'retry_until' was never true in {} retries: {}".format(
+                                        stage["name"], max_retries, retry_until
+                                    )
+                                ) from e
+                            elif isinstance(e, exceptions.TestFailError):
                                 raise
                             else:
                                 raise exceptions.TestFailError(
diff --git a/tests/integration/server.py b/tests/integration/server.py
index 31587609..6eced58f 100644
--- a/tests/integration/server.py
+++ b/tests/integration/server.py
@@ -344,6 +344,28 @@ def poll():
     return jsonify(response)
 
 
+job_polls: dict = {}
+
+
+@app.route("/job/<job_name>", methods=["GET"])
+def job_status(job_name):
+    """A long running job which is in progress for the first couple of polls and then
+    reaches a terminal state, which is 'FAILED' if the job name ends with 'fail'.
+
+    Once a job has finished it stays finished, so it can be polled again afterwards.
+    """
+    polls = job_polls[job_name] = job_polls.get(job_name, 0) + 1
+
+    if polls < 3:
+        status = "IN_PROGRESS"
+    elif job_name.endswith("fail"):
+        status = "FAILED"
+    else:
+        status = "SUCCESS"
+
+    return jsonify({"status": status})
+
+
 def _maybe_get_cookie_name():
     return (request.get_json(silent=True) or {}).get("cookie_name", "tavern-cookie")
 
diff --git a/tests/integration/starlark/README.md b/tests/integration/starlark/README.md
index b5811947..42ebce1d 100644
--- a/tests/integration/starlark/README.md
+++ b/tests/integration/starlark/README.md
@@ -35,3 +35,4 @@ docker-compose -f tests/integration/docker-compose.yml down
 ## Test Files
 
 - `test_control_flow_inline.tavern.yaml` - Basic pipeline test
+- `test_stage_conditions.tavern.yaml` - Per-stage `if`/`retry_until` expressions (no `control_flow` script)
diff --git a/tests/integration/starlark/test_stage_conditions.tavern.yaml b/tests/integration/starlark/test_stage_conditions.tavern.yaml
new file mode 100644
index 00000000..da0e1bcd
--- /dev/null
+++ b/tests/integration/starlark/test_stage_conditions.tavern.yaml
@@ -0,0 +1,373 @@
+is_defaults: True
+marks:
+  - starlark_control_flow
+
+---
+# Tests for the per-stage 'if' and 'retry_until' starlark expressions. These do not
+# need a 'control_flow' script, but do need the same experimental flag.
+
+test_name: Test per-stage 'if' with a saved variable
+
+stages:
+  - name: Echo a number to save
+    request:
+      url: "{global_host}/echo"
+      method: POST
+      json:
+        value: 5
+    response:
+      status_code: 200
+      save:
+        json:
+          var_x: value
+
+  - name: This stage should run
+    if: "{var_x} > 2"
+    request:
+      url: "{global_host}/echo"
+      method: POST
+      json:
+        value: "ran"
+    response:
+      status_code: 200
+      json:
+        value: "ran"
+
+  - name: This stage should not run
+    if: "{var_x} > 100"
+    request:
+      url: "{global_host}/echo"
+      method: POST
+      json:
+        value: "should not have run"
+    response:
+      # If this stage is ever actually run it will fail here
+      status_code: 999
+
+---
+test_name: Test per-stage 'if' using a nested value
+
+stages:
+  - name: Echo a dict to save
+    request:
+      url: "{global_host}/echo"
+      method: POST
+      json:
+        value:
+          status: ready
+    response:
+      status_code: 200
+      save:
+        json:
+          echoed: value
+
+  - name: This stage should run
+    if: "'{echoed.status}' == 'ready'"
+    request:
+      url: "{global_host}/echo"
+      method: POST
+      json:
+        value: "ran"
+    response:
+      status_code: 200
+      json:
+        value: "ran"
+
+---
+# Because expressions are format-string interpolated rather than bound as starlark
+# names, variables which aren't valid starlark identifiers work just as well
+test_name: Test per-stage 'if' using a variable with a dash in the name
+
+includes:
+  - name: dashed_vars
+    description: a variable which is not a valid starlark identifier
+    variables:
+      my-thing: 5
+
+stages:
+  - name: This stage should run
+    if: "{my-thing} > 4"
+    request:
+      url: "{global_host}/echo"
+      method: POST
+      json:
+        value: "ran"
+    response:
+      status_code: 200
+      json:
+        value: "ran"
+
+---
+# An 'expression' can be several statements long, with the value of the last one
+# deciding the result. The helper modules still have to be loaded, as in a
+# 'control_flow' script
+test_name: Test per-stage 'if' using a multiline script
+
+stages:
+  - name: Echo a version string to save
+    request:
+      url: "{global_host}/echo"
+      method: POST
+      json:
+        value: "server v2.5.1"
+    response:
+      status_code: 200
+      save:
+        json:
+          saved_version: value
+
+  - name: This stage should run because the major version is 2
+    if: |
+      load("@tavern_helpers.star", "re")
+
+      match = re.search("v(\\d+)\\.", "{saved_version}")
+      match != None and int(match.groups[0]) == 2
+    request:
+      url: "{global_host}/echo"
+      method: POST
+      json:
+        value: "ran"
+    response:
+      status_code: 200
+      json:
+        value: "ran"
+
+  - name: This stage should not run because the major version is not 3
+    if: |
+      load("@tavern_helpers.star", "re")
+
+      match = re.search("v(\\d+)\\.", "{saved_version}")
+      match != None and int(match.groups[0]) == 3
+    request:
+      url: "{global_host}/echo"
+      method: POST
+      json:
+        value: "should not have run"
+    response:
+      # If this stage is ever actually run it will fail here
+      status_code: 999
+
+---
+test_name: Test 'retry_until' using a multiline script
+
+stages:
+  - name: Poll until the job reaches a state matching a regex
+    request:
+      url: "{global_host}/job/multiline-works"
+      method: GET
+    response:
+      status_code: 200
+      json:
+        status: SUCCESS
+    max_retries: 5
+    delay_after: 0.1
+    retry_until: |
+      load("@tavern_helpers.star", "re")
+
+      terminal = re.match("(SUCCESS|FAILED)", response.body["status"])
+      terminal != None
+
+---
+test_name: Test per-stage 'if' referring to a variable that was never saved
+
+_xfail: run
+
+stages:
+  - name: This stage errors because the variable is not defined
+    if: "{never_saved} == 1"
+    request:
+      url: "{global_host}/echo"
+      method: POST
+      json:
+        value: "hello"
+    response:
+      status_code: 200
+
+---
+test_name: Test 'retry_until' polling until the response body is ready
+
+stages:
+  # The response block only matches when the poll endpoint says 'ready'. Until then the
+  # stage fails, and 'retry_until' is consulted to decide whether to keep going.
+  - name: Poll until ready
+    request:
+      url: "{global_host}/poll"
+      method: GET
+    response:
+      status_code: 200
+      json:
+        status: ready
+    max_retries: 5
+    delay_after: 0.1
+    retry_until: response.body["status"] == "ready"
+
+---
+test_name: Test 'retry_until' is not evaluated when the stage passes
+
+stages:
+  # 'retry_until' can never be true, but the stage itself passes first time, so it is
+  # never evaluated and the stage is not retried
+  - name: Poll once
+    request:
+      url: "{global_host}/poll"
+      method: GET
+    response:
+      status_code: 200
+    max_retries: 2
+    delay_after: 0.1
+    retry_until: response.body["status"] == "never-going-to-happen"
+
+---
+test_name: Test 'retry_until' which is never true fails the test
+
+_xfail: run
+
+stages:
+  - name: Poll for something that never happens
+    request:
+      url: "{global_host}/poll"
+      method: GET
+    response:
+      # Never matches, so the stage always fails and retry_until is always consulted
+      status_code: 418
+    max_retries: 2
+    delay_after: 0.1
+    retry_until: response.body["status"] == "never-going-to-happen"
+
+---
+# https://github.com/taverntesting/tavern/issues/751 - poll a long running job until it
+# reaches _any_ terminal state, rather than guessing how many retries it will need
+test_name: Test 'retry_until' stopping on any terminal state
+
+stages:
+  - name: Poll until the job finishes
+    request:
+      url: "{global_host}/job/works"
+      method: GET
+    response:
+      status_code: 200
+      json:
+        status: SUCCESS
+    max_retries: 5
+    delay_after: 0.1
+    retry_until: response.body["status"] == "SUCCESS" or response.body["status"] == "FAILED"
+
+---
+test_name: Test 'retry_until' stopping on a terminal state which is a failure
+
+stages:
+  # The job reaches 'FAILED', which is still finished as far as the polling is
+  # concerned, so 'retry_until' stops the retries even though this stage fails
+  - name: Poll until the job finishes
+    request:
+      url: "{global_host}/job/doomed-to-fail"
+      method: GET
+    response:
+      status_code: 200
+      json:
+        status: SUCCESS
+    max_retries: 5
+    delay_after: 0.1
+    retry_until: response.body["status"] == "SUCCESS" or response.body["status"] == "FAILED"
+
+  # Because a stage which finished via 'retry_until' does not save anything, check what
+  # the job actually ended up as in a separate stage
+  - name: Check the job failed
+    request:
+      url: "{global_host}/job/doomed-to-fail"
+      method: GET
+    response:
+      status_code: 200
+      json:
+        status: FAILED
+
+---
+test_name: Test 'retry_until' can use the status code and test variables
+
+includes:
+  - name: retry_until_vars
+    description: variables used in the retry_until expression
+    variables:
+      expected_status: ready
+
+stages:
+  - name: Poll until ready
+    request:
+      url: "{global_host}/poll"
+      method: GET
+    response:
+      status_code: 418
+    max_retries: 5
+    delay_after: 0.1
+    retry_until: response.status_code == 200 and response.body["status"] == "{expected_status}"
+
+---
+test_name: Test 'fail_if' as a negative assertion on a stage which otherwise passes
+
+_xfail: run
+
+stages:
+  # The response block matches, but 'fail_if' says this is a failure anyway
+  - name: Echo something which should never come back
+    request:
+      url: "{global_host}/echo"
+      method: POST
+      json:
+        value: "a bad value"
+    response:
+      status_code: 200
+    fail_if: response.body["value"] == "a bad value"
+
+---
+test_name: Test 'fail_if' which is false does not affect the stage
+
+stages:
+  - name: Echo something fine
+    request:
+      url: "{global_host}/echo"
+      method: POST
+      json:
+        value: "a good value"
+    response:
+      status_code: 200
+      json:
+        value: "a good value"
+    fail_if: response.body["value"] == "a bad value"
+
+---
+# https://github.com/taverntesting/tavern/issues/751 - stop polling as soon as the job
+# is in a state it can never recover from, rather than waiting out all the retries
+test_name: Test 'fail_if' stops polling at a terminal failure
+
+_xfail: run
+
+stages:
+  - name: Poll until the job succeeds
+    request:
+      url: "{global_host}/job/fail-if-doomed-to-fail"
+      method: GET
+    response:
+      status_code: 200
+      json:
+        status: SUCCESS
+    max_retries: 20
+    delay_after: 0.1
+    retry_until: response.body["status"] == "SUCCESS"
+    fail_if: response.body["status"] == "FAILED"
+
+---
+test_name: Test 'fail_if' does not trigger for a job which succeeds
+
+stages:
+  - name: Poll until the job succeeds
+    request:
+      url: "{global_host}/job/fail-if-works"
+      method: GET
+    response:
+      status_code: 200
+      json:
+        status: SUCCESS
+    max_retries: 20
+    delay_after: 0.1
+    retry_until: response.body["status"] == "SUCCESS"
+    fail_if: response.body["status"] == "FAILED"
diff --git a/tests/unit/starlark/test_expressions.py b/tests/unit/starlark/test_expressions.py
new file mode 100644
index 00000000..e2e7f114
--- /dev/null
+++ b/tests/unit/starlark/test_expressions.py
@@ -0,0 +1,230 @@
+import dataclasses
+
+import pytest
+
+from tavern._core import exceptions
+from tavern._core.starlark.expressions import eval_expression, eval_stage_expression
+
+
+class TestEvalExpression:
+    def test_simple_true(self):
+        assert eval_expression("1 < 2", {}, description="test") is True
+
+    def test_simple_false(self):
+        assert eval_expression("1 > 2", {}, description="test") is False
+
+    def test_variable_interpolated(self):
+        """Variables are interpolated into the expression before it is evaluated"""
+        assert eval_expression("{var_x} > 2", {"var_x": 3}, description="test") is True
+        assert eval_expression("{var_x} > 2", {"var_x": 1}, description="test") is False
+
+    def test_string_variable(self):
+        assert (
+            eval_expression(
+                "'{some_var}' == 'value'", {"some_var": "value"}, description="test"
+            )
+            is True
+        )
+
+    def test_nested_variable(self):
+        assert (
+            eval_expression(
+                "{thing.a.b} == 1",
+                {"thing": {"a": {"b": 1}}},
+                description="test",
+            )
+            is True
+        )
+
+    def test_variable_with_a_dash(self):
+        """Names which aren't valid starlark identifiers work fine when interpolated"""
+        assert (
+            eval_expression("{with-a-dash} > 4", {"with-a-dash": 5}, description="test")
+            is True
+        )
+
+    def test_format_spec(self):
+        assert (
+            eval_expression(
+                "'{my_float:.2f}' == '1.50'", {"my_float": 1.5}, description="test"
+            )
+            is True
+        )
+
+    def test_undefined_variable(self):
+        with pytest.raises(exceptions.EvalError) as exc_info:
+            eval_expression("{not_a_variable} > 1", {}, description="test")
+
+        assert "not_a_variable" in str(exc_info.value)
+
+    def test_undefined_bare_name(self):
+        """A bare name which isn't interpolated is just undefined in starlark"""
+        with pytest.raises(exceptions.EvalError) as exc_info:
+            eval_expression("not_a_variable", {}, description="test")
+
+        assert "not_a_variable" in str(exc_info.value)
+
+    def test_invalid_syntax(self):
+        with pytest.raises(exceptions.EvalError):
+            eval_expression("hello i am a test <<<", {}, description="test")
+
+    def test_non_bool_result(self):
+        with pytest.raises(exceptions.EvalError) as exc_info:
+            eval_expression("'a string'", {}, description="test")
+
+        assert "did not evaluate to True/False" in str(exc_info.value)
+
+    def test_reserved_word_variables(self):
+        """Names which are reserved words in starlark are fine when interpolated"""
+        assert (
+            eval_expression("{load} == 1", {"load": 1, "fine": 3}, description="test")
+            is True
+        )
+
+    def test_unreferenced_variables_are_ignored(self):
+        """Variables which aren't referenced shouldn't break anything, whatever they are"""
+
+        class Something:
+            pass
+
+        variables = {
+            "opaque": Something(),
+            "1_starts_with_number": 2,
+            "fine": 3,
+        }
+
+        assert eval_expression("{fine} == 3", variables, description="test") is True
+
+    def test_literal_braces_must_be_escaped(self):
+        assert eval_expression("{{'a': 1}}['a'] == 1", {}, description="test") is True
+
+    def test_response_struct_attribute_access(self):
+        response = {"status_code": 200, "body": {"status": "ready"}, "failed": False}
+
+        assert (
+            eval_expression(
+                "response.status_code == 200 and response.body['status'] == 'ready'",
+                {},
+                response=response,
+                description="test",
+            )
+            is True
+        )
+
+    def test_response_and_variables_together(self):
+        response = {"status_code": 500}
+
+        assert (
+            eval_expression(
+                "response.status_code == {expected_code}",
+                {"expected_code": 500},
+                response=response,
+                description="test",
+            )
+            is True
+        )
+
+    def test_response_missing_key(self):
+        with pytest.raises(exceptions.EvalError):
+            eval_expression(
+                "response.nonexistent == 1",
+                {},
+                response={"status_code": 200},
+                description="test",
+            )
+
+
+class TestMultilineExpression:
+    """An 'expression' can be several statements, the last of which is the result"""
+
+    def test_last_statement_is_the_result(self):
+        script = """
+x = {var_x} * 2
+y = x + 1
+y == 7
+"""
+        assert eval_expression(script, {"var_x": 3}, description="test") is True
+
+    def test_last_statement_must_be_a_boolean(self):
+        script = """
+x = 1
+x
+"""
+        with pytest.raises(exceptions.EvalError) as exc_info:
+            eval_expression(script, {}, description="test")
+
+        assert "did not evaluate to True/False" in str(exc_info.value)
+
+    def test_ending_with_an_assignment_is_an_error(self):
+        """Only an expression statement has a value - anything else is None, which is
+        not a useful answer to 'should this stage run'"""
+        script = """
+x = 1
+result = x == 1
+"""
+        with pytest.raises(exceptions.EvalError) as exc_info:
+            eval_expression(script, {}, description="test")
+
+        assert "did not evaluate to True/False" in str(exc_info.value)
+
+    def test_can_load_the_regex_helpers(self):
+        script = r"""
+load("@tavern_helpers.star", "re")
+
+match = re.search("v(\\d+)\\.", response.body["version"])
+match != None and match.groups[0] == "{expected_major}"
+"""
+        assert (
+            eval_expression(
+                script,
+                {"expected_major": 25},
+                response={"body": {"version": "v25.3.1"}},
+                description="test",
+            )
+            is True
+        )
+
+    def test_load_of_something_else_is_an_error(self):
+        with pytest.raises(exceptions.EvalError):
+            eval_expression(
+                'load("@not_a_module.star", "thing")\nTrue',
+                {},
+                description="test",
+            )
+
+    def test_run_stage_is_not_available(self):
+        script = """
+load("@tavern_helpers.star", "run_stage")
+
+run_stage("some_stage").failed
+"""
+        with pytest.raises(exceptions.EvalError) as exc_info:
+            eval_expression(script, {}, description="test")
+
+        assert "control_flow" in str(exc_info.value)
+
+
+class TestStageExpressionGuard:
+    def test_requires_experimental_flag(self, fix_test_config):
+        config = dataclasses.replace(
+            fix_test_config, experimental_starlark_pipeline=False
+        )
+
+        with pytest.raises(exceptions.UnexpectedKeysError) as exc_info:
+            eval_stage_expression("if", "1 < 2", {"name": "a stage"}, config)
+
+        assert "--tavern-experimental-starlark-pipeline" in str(exc_info.value)
+
+    def test_works_with_experimental_flag(self, fix_test_config):
+        assert (
+            eval_stage_expression("if", "1 < 2", {"name": "a stage"}, fix_test_config)
+            is True
+        )
+
+    def test_stage_name_in_error(self, fix_test_config):
+        with pytest.raises(exceptions.EvalError) as exc_info:
+            eval_stage_expression(
+                "if", "not_a_variable", {"name": "a stage"}, fix_test_config
+            )
+
+        assert "a stage" in str(exc_info.value)
diff --git a/tests/unit/starlark/test_starlark_env.py b/tests/unit/starlark/test_starlark_env.py
index 1a57245e..a47f3b96 100644
--- a/tests/unit/starlark/test_starlark_env.py
+++ b/tests/unit/starlark/test_starlark_env.py
@@ -12,11 +12,11 @@
 
 from tavern._core import exceptions
 from tavern._core.run import _TestRunner
+from tavern._core.starlark.builtins import wrap_callable
 from tavern._core.starlark.stage_registry import StageRegistry
 from tavern._core.starlark.starlark_env import (
     StageResponse,
     StarlarkPipelineRunner,
-    _wrap_callable,
 )
 from tavern._core.tincture import Tinctures
 
@@ -57,12 +57,12 @@ def sample_stages():
 
 
 class TestWrapCallable:
-    """Tests for the _wrap_callable decorator."""
+    """Tests for the wrap_callable decorator."""
 
     def test_wrap_callable_converts_args_to_starlark(self):
-        """Test that _wrap_callable converts Python args to starlark format."""
+        """Test that wrap_callable converts Python args to starlark format."""
 
-        @_wrap_callable
+        @wrap_callable
         def add(a, b):
             return a + b
 
@@ -71,9 +71,9 @@ def add(a, b):
         assert result == 3
 
     def test_wrap_callable_converts_kwargs_to_starlark(self):
-        """Test that _wrap_callable converts Python kwargs to starlark format."""
+        """Test that wrap_callable converts Python kwargs to starlark format."""
 
-        @_wrap_callable
+        @wrap_callable
         def format_url(base, path=""):
             return f"{base}{path}"
 
@@ -81,9 +81,9 @@ def format_url(base, path=""):
         assert result == "http://example.com/api"
 
     def test_wrap_callable_converts_return_to_starlark(self):
-        """Test that _wrap_callable converts return value to starlark format."""
+        """Test that wrap_callable converts return value to starlark format."""
 
-        @_wrap_callable
+        @wrap_callable
         def get_dict():
             return {"key": "value"}
 
@@ -91,12 +91,12 @@ def get_dict():
         assert result == {"key": "value"}
 
     def test_wrap_callable_converts_opaque_return_to_starlark(self):
-        """Test that _wrap_callable converts opaque return value to starlark format."""
+        """Test that wrap_callable converts opaque return value to starlark format."""
 
         class _boobllb:
             pass
 
-        @_wrap_callable
+        @wrap_callable
         def get_dict():
             return _boobllb
 
diff --git a/tests/unit/test_schema.py b/tests/unit/test_schema.py
index 3cedd9e5..ed3710cb 100644
--- a/tests/unit/test_schema.py
+++ b/tests/unit/test_schema.py
@@ -133,6 +133,56 @@ def test_verify_with_incorrect_value(self, test_dict, incorrect_value):
             verify_tests(test_dict)
 
 
+class TestStageConditions:
+    """The per-stage starlark 'if'/'retry_until' keys"""
+
+    def test_if_alone(self, test_dict):
+        test_dict["stages"][0]["if"] = "var_x > 2"
+        verify_tests(test_dict)
+
+    def test_if_and_skip_together(self, test_dict):
+        """'if' is the starlark equivalent of 'skip' so using both is nonsense"""
+        test_dict["stages"][0]["if"] = "var_x > 2"
+        test_dict["stages"][0]["skip"] = "True"
+
+        with pytest.raises(BadSchemaError):
+            verify_tests(test_dict)
+
+    def test_retry_until_with_max_retries(self, test_dict):
+        test_dict["stages"][0]["retry_until"] = "response.status_code == 200"
+        test_dict["stages"][0]["max_retries"] = 3
+        verify_tests(test_dict)
+
+    def test_retry_until_without_max_retries(self, test_dict):
+        test_dict["stages"][0]["retry_until"] = "response.status_code == 200"
+
+        with pytest.raises(BadSchemaError):
+            verify_tests(test_dict)
+
+    def test_if_must_be_a_string(self, test_dict):
+        test_dict["stages"][0]["if"] = True
+
+        with pytest.raises(BadSchemaError):
+            verify_tests(test_dict)
+
+    def test_fail_if_alone(self, test_dict):
+        """Unlike 'retry_until', 'fail_if' does not need any retries to be useful"""
+        test_dict["stages"][0]["fail_if"] = "response.status_code == 500"
+        verify_tests(test_dict)
+
+    def test_fail_if_with_retry_until(self, test_dict):
+        test_dict["stages"][0]["fail_if"] = "response.body['status'] == 'FAILED'"
+        test_dict["stages"][0]["retry_until"] = "response.body['status'] == 'SUCCESS'"
+        test_dict["stages"][0]["max_retries"] = 3
+        verify_tests(test_dict)
+
+    def test_fail_if_must_be_a_string(self, test_dict):
+        test_dict["stages"][0]["fail_if"] = True
+
+        with pytest.raises(BadSchemaError):
+            verify_tests(test_dict)
+
+
 class TestBadSchemaAtCollect:
     """Some errors happen at collection time - harder to test"""
 
diff --git a/tests/unit/test_stage_conditions.py b/tests/unit/test_stage_conditions.py
new file mode 100644
index 00000000..bb660eb5
--- /dev/null
+++ b/tests/unit/test_stage_conditions.py
@@ -0,0 +1,455 @@
+"""Tests for the per-stage 'if', 'retry_until' and 'fail_if' starlark expressions"""
+
+import dataclasses
+import unittest.mock
+from collections.abc import Mapping
+from unittest.mock import Mock, create_autospec, patch
+
+import pytest
+import requests
+
+from tavern._core import exceptions
+from tavern._core.pytest.config import TestConfig
+from tavern._core.run import _TestRunner, run_test
+from tavern._core.strict_util import StrictLevel
+from tavern._core.testhelpers import retry
+from tavern._core.tincture import Tinctures
+from tavern.request import BaseRequest
+from tavern.response import BaseResponse
+
+
+def _run_test(
+    stage: Mapping, test_block_config: TestConfig, run_mock: unittest.mock.Mock
+) -> bool:
+    """runs the test and returns whether the stage was run or not"""
+
+    full_test = {
+        "test_name": "A test with a single stage",
+        "stages": [stage],
+    }
+
+    run_test("test_file_name", full_test, test_block_config)
+
+    return run_mock.called
+
+
+class TestIfStage:
+    @pytest.fixture(autouse=True)
+    def run_mock(self):
+        with patch("tavern._core.run._TestRunner.run_stage") as run_mock:
+            yield run_mock
+
+    @pytest.fixture(scope="function")
+    def stage(self):
+        return {
+            "name": "test stage",
+            "request": {"url": "https://example.com", "method": "GET"},
+            "response": {"status_code": 200},
+        }
+
+    @pytest.fixture
+    def test_block_config(self, includes):
+        return dataclasses.replace(
+            includes,
+            variables={"env_vars": {}},
+            experimental_starlark_pipeline=True,
+        )
+
+    def test_if_true_runs_stage(self, stage, test_block_config, run_mock):
+        stage["if"] = "1 < 2"
+        assert _run_test(stage, test_block_config, run_mock) is True
+
+    def test_if_false_skips_stage(self, stage, test_block_config, run_mock):
+        stage["if"] = "1 > 2"
+        assert _run_test(stage, test_block_config, run_mock) is False
+
+    def test_if_uses_saved_variable(self, stage, test_block_config, run_mock):
+        stage["if"] = "{var_x} > 2"
+        test_block_config.variables.update({"var_x": 3})
+
+        assert _run_test(stage, test_block_config, run_mock) is True
+
+    def test_if_uses_saved_variable_false(self, stage, test_block_config, run_mock):
+        stage["if"] = "{var_x} > 2"
+        test_block_config.variables.update({"var_x": 1})
+
+        assert _run_test(stage, test_block_config, run_mock) is False
+
+    def test_if_undefined_variable(self, stage, test_block_config, run_mock):
+        stage["if"] = "{not_saved_yet} == 1"
+
+        with pytest.raises(exceptions.EvalError):
+            _run_test(stage, test_block_config, run_mock)
+
+    def test_if_non_bool_result(self, stage, test_block_config, run_mock):
+        stage["if"] = "'a string'"
+
+        with pytest.raises(exceptions.EvalError):
+            _run_test(stage, test_block_config, run_mock)
+
+    def test_if_requires_experimental_flag(self, stage, test_block_config, run_mock):
+        stage["if"] = "1 < 2"
+        test_block_config = dataclasses.replace(
+            test_block_config, experimental_starlark_pipeline=False
+        )
+
+        with pytest.raises(exceptions.UnexpectedKeysError):
+            _run_test(stage, test_block_config, run_mock)
+
+
+def _mock_response(body, status_code=200):
+    response = Mock(spec=requests.Response)
+    response.status_code = status_code
+    response.headers = {"Content-Type": "application/json"}
+    response.json.return_value = body
+    response.cookies = {}
+    response.content = b"{}"
+    return response
+
+
+def _stage_failure(body, status_code=200):
+    """A stage which failed verification, but which did get a response back"""
+    error = exceptions.TestFailError("stage did not verify")
+    error.response = _mock_response(body, status_code)
+    return error
+
+
+class TestRetryUntil:
+    @pytest.fixture
+    def test_block_config(self, includes):
+        return dataclasses.replace(
+            includes,
+            variables={"env_vars": {}},
+            experimental_starlark_pipeline=True,
+        )
+
+    @pytest.fixture
+    def stage(self):
+        return {
+            "name": "test stage",
+            "max_retries": 3,
+            "retry_until": "response.body['status'] == 'ready'",
+        }
+
+    def test_not_evaluated_when_stage_passes(self, stage, test_block_config):
+        """A stage which passes is finished - retry_until is not consulted at all,
+        even though it would have been false"""
+        response = _mock_response({"status": "pending"})
+        inner = Mock(return_value=response)
+
+        assert retry(stage, test_block_config)(inner)() is response
+        assert inner.call_count == 1
+
+    def test_stops_when_retry_until_is_true(self, stage, test_block_config):
+        """The stage keeps failing, but retry_until eventually becomes true"""
+        failures = [
+            _stage_failure({"status": "pending"}),
+            _stage_failure({"status": "pending"}),
+            _stage_failure({"status": "ready"}),
+        ]
+        inner = Mock(side_effect=failures)
+
+        assert retry(stage, test_block_config)(inner)() is failures[-1].response
+        assert inner.call_count == 3
+
+    def test_stops_immediately_when_retry_until_is_true(self, stage, test_block_config):
+        inner = Mock(side_effect=_stage_failure({"status": "ready"}))
+
+        retry(stage, test_block_config)(inner)()
+        assert inner.call_count == 1
+
+    def test_fails_after_exhausting_retries(self, stage, test_block_config):
+        inner = Mock(
+            side_effect=lambda: (_ for _ in ()).throw(
+                _stage_failure({"status": "pending"})
+            )
+        )
+
+        with pytest.raises(exceptions.TestFailError) as exc_info:
+            retry(stage, test_block_config)(inner)()
+
+        # max_retries = 3 means 4 attempts in total
+        assert inner.call_count == 4
+        assert "retry_until" in str(exc_info.value)
+
+    @pytest.mark.parametrize("terminal_status", ["SUCCESS", "FAILED"])
+    def test_stops_on_any_terminal_state(
+        self, stage, test_block_config, terminal_status
+    ):
+        """Poll a long running job until it finishes, whether it succeeded or not
+
+        https://github.com/taverntesting/tavern/issues/751
+        """
+        stage["retry_until"] = (
+            "response.body['status'] == 'SUCCESS'"
+            " or response.body['status'] == 'FAILED'"
+        )
+        stage["max_retries"] = 5
+        failures = [
+            _stage_failure({"status": "IN_PROGRESS"}),
+            _stage_failure({"status": "IN_PROGRESS"}),
+            _stage_failure({"status": terminal_status}),
+        ]
+        inner = Mock(side_effect=failures)
+
+        assert retry(stage, test_block_config)(inner)() is failures[-1].response
+        assert inner.call_count == 3
+
+    def test_never_reaching_a_terminal_state_fails(self, stage, test_block_config):
+        stage["retry_until"] = (
+            "response.body['status'] == 'SUCCESS'"
+            " or response.body['status'] == 'FAILED'"
+        )
+        stage["max_retries"] = 2
+        inner = Mock(
+            side_effect=lambda: (_ for _ in ()).throw(
+                _stage_failure({"status": "IN_PROGRESS"})
+            )
+        )
+
+        with pytest.raises(exceptions.TestFailError):
+            retry(stage, test_block_config)(inner)()
+
+        assert inner.call_count == 3
+
+    def test_not_evaluated_without_a_response(self, stage, test_block_config):
+        """If the request itself failed there is no response to inspect, so just retry"""
+        inner = Mock(
+            side_effect=[
+                exceptions.TestFailError("no response at all"),
+                _mock_response({"status": "pending"}),
+            ]
+        )
+
+        retry(stage, test_block_config)(inner)()
+        assert inner.call_count == 2
+
+    def test_delay_after_between_attempts(self, stage, test_block_config):
+        stage["delay_after"] = 0.01
+        inner = Mock(
+            side_effect=[
+                _stage_failure({"status": "pending"}),
+                _stage_failure({"status": "ready"}),
+            ]
+        )
+
+        with patch("tavern._core.testhelpers.time.sleep") as sleep_mock:
+            retry(stage, test_block_config)(inner)()
+
+        sleep_mock.assert_called_once_with(0.01)
+
+    def test_uses_test_variables(self, stage, test_block_config):
+        stage["retry_until"] = "response.body['status'] == '{expected_status}'"
+        test_block_config.variables["expected_status"] = "ready"
+        inner = Mock(side_effect=_stage_failure({"status": "ready"}))
+
+        retry(stage, test_block_config)(inner)()
+        assert inner.call_count == 1
+
+    def test_uses_status_code(self, stage, test_block_config):
+        stage["retry_until"] = "response.status_code == 201"
+        inner = Mock(
+            side_effect=[
+                _stage_failure({}, status_code=503),
+                _stage_failure({}, status_code=201),
+            ]
+        )
+
+        retry(stage, test_block_config)(inner)()
+        assert inner.call_count == 2
+
+    def test_without_max_retries_is_an_error(self, stage, test_block_config):
+        del stage["max_retries"]
+
+        with pytest.raises(exceptions.InvalidRetryException):
+            retry(stage, test_block_config)
+
+    def test_requires_experimental_flag(self, stage, test_block_config):
+        test_block_config = dataclasses.replace(
+            test_block_config, experimental_starlark_pipeline=False
+        )
+        inner = Mock(side_effect=_stage_failure({"status": "ready"}))
+
+        with pytest.raises(exceptions.UnexpectedKeysError):
+            retry(stage, test_block_config)(inner)()
+
+
+def _stage_callable():
+    """The signature of the function that 'retry' wraps, used as a Mock spec"""
+
+
+def _mock_stage_callable(**kwargs) -> Mock:
+    """A mock of the function that 'retry' wraps
+
+    This is autospecced rather than a plain Mock because the retry wrapper calls
+    functools.wraps on it, which needs the real function attributes.
+    """
+    return create_autospec(_stage_callable, **kwargs)
+
+
+def _run_stage(stage, test_block_config, response, *, verify_error=None):
+    """Run a single stage, mocking out everything to do with actually making a request
+
+    Args:
+        stage: the stage to run
+        test_block_config: config for the test
+        response: what the 'request' should return
+        verify_error: if given, an exception raised when verifying the response
+    """
+
+    verifier = Mock(spec=BaseResponse)
+    if verify_error is not None:
+        verifier.verify.side_effect = verify_error
+    else:
+        verifier.verify.return_value = {}
+
+    request = Mock(spec=BaseRequest)
+    request.request_vars = {}
+    request.run.return_value = response
+
+    runner = _TestRunner(
+        default_global_strictness=StrictLevel.all_on(),
+        sessions={},
+        test_block_config=test_block_config,
+        test_spec={"test_name": "a test", "stages": [stage]},
+    )
+
+    with (
+        patch("tavern._core.run.attach_stage_content"),
+        patch("tavern._core.run.call_hook"),
+        patch("tavern._core.run.get_request_type", return_value=request),
+        patch("tavern._core.run.get_expected", return_value={}),
+        patch("tavern._core.run.get_verifiers", return_value={"response": [verifier]}),
+    ):
+        return runner.wrapped_run_stage(stage, test_block_config, Mock(spec=Tinctures))
+
+
+class TestFailIf:
+    @pytest.fixture
+    def test_block_config(self, includes):
+        return dataclasses.replace(
+            includes,
+            variables={"env_vars": {}, "tavern": {}},
+            experimental_starlark_pipeline=True,
+        )
+
+    @pytest.fixture
+    def stage(self):
+        return {
+            "name": "test stage",
+            "request": {"url": "https://example.com", "method": "GET"},
+            "response": {"status_code": 200},
+            "fail_if": "response.body['status'] == 'FAILED'",
+        }
+
+    def test_passing_stage_with_false_expression(self, stage, test_block_config):
+        response = _mock_response({"status": "SUCCESS"})
+
+        assert _run_stage(stage, test_block_config, response) is response
+
+    def test_passing_stage_with_true_expression(self, stage, test_block_config):
+        """The response block matched, but the stage is a failure anyway"""
+        response = _mock_response({"status": "FAILED"})
+
+        with pytest.raises(exceptions.FailIfError) as exc_info:
+            _run_stage(stage, test_block_config, response)
+
+        assert "fail_if" in str(exc_info.value)
+
+    def test_failing_stage_with_true_expression(self, stage, test_block_config):
+        response = _mock_response({"status": "FAILED"})
+
+        with pytest.raises(exceptions.FailIfError):
+            _run_stage(
+                stage,
+                test_block_config,
+                response,
+                verify_error=exceptions.TestFailError("stage did not verify"),
+            )
+
+    def test_failing_stage_with_false_expression(self, stage, test_block_config):
+        """The normal failure is unaffected by a 'fail_if' which was false"""
+        response = _mock_response({"status": "IN_PROGRESS"})
+
+        with pytest.raises(exceptions.TestFailError) as exc_info:
+            _run_stage(
+                stage,
+                test_block_config,
+                response,
+                verify_error=exceptions.TestFailError("stage did not verify"),
+            )
+
+        assert not isinstance(exc_info.value, exceptions.FailIfError)
+
+    def test_uses_test_variables(self, stage, test_block_config):
+        stage["fail_if"] = "response.body['status'] == '{bad_status}'"
+        test_block_config.variables["bad_status"] = "FAILED"
+
+        with pytest.raises(exceptions.FailIfError):
+            _run_stage(stage, test_block_config, _mock_response({"status": "FAILED"}))
+
+    def test_knows_the_stage_failed(self, stage, test_block_config):
+        stage["fail_if"] = "response.failed"
+
+        with pytest.raises(exceptions.FailIfError):
+            _run_stage(
+                stage,
+                test_block_config,
+                _mock_response({"status": "SUCCESS"}),
+                verify_error=exceptions.TestFailError("stage did not verify"),
+            )
+
+    def test_knows_the_stage_passed(self, stage, test_block_config):
+        stage["fail_if"] = "response.failed"
+        response = _mock_response({"status": "SUCCESS"})
+
+        assert _run_stage(stage, test_block_config, response) is response
+
+    def test_non_bool_result(self, stage, test_block_config):
+        stage["fail_if"] = "response.body['status']"
+
+        with pytest.raises(exceptions.EvalError):
+            _run_stage(stage, test_block_config, _mock_response({"status": "FAILED"}))
+
+    def test_must_be_a_string(self, stage, test_block_config):
+        stage["fail_if"] = True
+
+        with pytest.raises(exceptions.BadSchemaError):
+            _run_stage(stage, test_block_config, _mock_response({"status": "FAILED"}))
+
+    def test_requires_experimental_flag(self, stage, test_block_config):
+        test_block_config = dataclasses.replace(
+            test_block_config, experimental_starlark_pipeline=False
+        )
+
+        with pytest.raises(exceptions.UnexpectedKeysError):
+            _run_stage(stage, test_block_config, _mock_response({"status": "FAILED"}))
+
+    def test_is_not_retried(self, stage, test_block_config):
+        """A stage which hit its 'fail_if' is in a terminal state, so don't retry it
+
+        https://github.com/taverntesting/tavern/issues/751
+        """
+        stage["max_retries"] = 5
+        inner = _mock_stage_callable(
+            side_effect=exceptions.FailIfError("fail_if was true")
+        )
+
+        with pytest.raises(exceptions.FailIfError):
+            retry(stage, test_block_config)(inner)()
+
+        assert inner.call_count == 1
+
+    def test_takes_priority_over_retry_until(self, stage, test_block_config):
+        """Both keys are evaluated, but 'fail_if' is checked while running the stage so
+        it never reaches the 'retry_until' handling in the retry wrapper"""
+        stage["max_retries"] = 5
+        stage["retry_until"] = "True"
+        inner = _mock_stage_callable(
+            side_effect=exceptions.FailIfError("fail_if was true")
+        )
+
+        with pytest.raises(exceptions.FailIfError):
+            retry(stage, test_block_config)(inner)()
+
+        assert inner.call_count == 1