From 5d4ba64aee55ef2a2a83c48a5751d2d653e91d5a Mon Sep 17 00:00:00 2001
From: openhands <openhands@all-hands.dev>
Date: Wed, 21 Oct 2026 18:04:34 +0000
Subject: [PATCH 1/4] feat: Spec getExperimentFlag, trackEvent and
 trackExposureEvent

Adds the `@events` capability: three feature files covering the SDK methods
that power experimentation, and the pieces the runner needs to check them.

SDKs send tracking and exposure events to Flagsmith's events API, which the
open-source Flagsmith image does not serve, so `events-api/` stands in for it
and records what an SDK sends. Events are off by default in every SDK and its
URL is not derived from the Flagsmith API URL, so scenarios that need them say
`Given the SDK sends events to Flagsmith` and the runner creates the client
with `Create the SDK client with events enabled` instead. SDKs that have not
implemented that step report it missing, as for any other new step.

Every harness now recognises a step marker starting with `Create the SDK
client`, rather than only that exact text. Existing steps files are unaffected.

Closes #13
---
 HARNESS.md                                    |  8 +-
 README.md                                     | 23 ++++-
 compose.yml                                   | 15 ++++
 events-api/events_api.py                      | 77 ++++++++++++++++
 features/get_experiment_flag.feature          | 38 ++++++++
 features/track_event.feature                  | 17 ++++
 features/track_exposure_event.feature         | 16 ++++
 harness/dart/README.md                        |  2 +-
 harness/dart/build.dart                       |  2 +-
 harness/dotnet/Harness.cs                     |  2 +-
 harness/elixir/harness.exs                    |  2 +-
 harness/go/build.go                           |  2 +-
 harness/java/Harness.java                     |  2 +-
 harness/kotlin/Harness.kt                     |  2 +-
 harness/nodejs/harness.mts                    |  2 +-
 harness/php/harness.php                       |  2 +-
 harness/python/harness.py                     |  2 +-
 harness/ruby/harness.rb                       |  2 +-
 harness/rust/README.md                        |  2 +-
 harness/rust/build.rs                         |  6 +-
 harness/swift/README.md                       |  2 +-
 .../Sources/StepsTable/StepsTable.swift       |  4 +-
 .../StepsTableTests/StepsTableTests.swift     | 19 ++++
 runner/pyproject.toml                         |  1 +
 runner/sdk_spec/events_api.py                 | 40 +++++++++
 runner/tests/conftest.py                      | 88 ++++++++++++++++++-
 26 files changed, 350 insertions(+), 28 deletions(-)
 create mode 100644 events-api/events_api.py
 create mode 100644 features/get_experiment_flag.feature
 create mode 100644 features/track_event.feature
 create mode 100644 features/track_exposure_event.feature
 create mode 100644 runner/sdk_spec/events_api.py

diff --git a/HARNESS.md b/HARNESS.md
index 7ed07f4..6f13cd7 100644
--- a/HARNESS.md
+++ b/HARNESS.md
@@ -20,10 +20,12 @@ A harness runs an SDK's steps file for the spec runner. There is one per languag
 
 ## Finding steps
 
-- A step marker is a line comment holding either `Create the SDK client` or a step starting with
-  `When `, e.g. `When the application gets the environment flags`.
+- A step marker is a line comment starting with either `Create the SDK client` or `When `, e.g.
+  `When the application gets the environment flags`.
 - The marked function is the first function defined after the marker.
-- The marker text identifies the step, and must match the runner's text exactly.
+- The marker text identifies the step, and must match the runner's text exactly. A steps file has
+  one function per step, including the client-creating steps the runner asks for by name
+  (`Create the SDK client`, `Create the SDK client with events enabled`).
 
 Find steps at runtime where the language allows it. Otherwise, generate a dispatch table at build time
 inside the SDK's Docker build (Go does this, for example).
diff --git a/README.md b/README.md
index be276c0..fa667d5 100644
--- a/README.md
+++ b/README.md
@@ -7,6 +7,7 @@ executed against every SDK.
 - [`runner/`](runner) — executes the spec (pytest-bdd). Step definitions live here, written once for
   every SDK.
 - [`compose.yml`](compose.yml) — the Flagsmith API (pinned) the spec runs against.
+- [`events-api/`](events-api) — Flagsmith's events API, as far as the spec is concerned.
 - [`harness/`](harness) — one harness per language, running an SDK's steps file
   ([`HARNESS.md`](HARNESS.md)).
 - [`.github/workflows/run-spec.yml`](.github/workflows/run-spec.yml) — reusable workflow SDKs call.
@@ -20,9 +21,12 @@ executed against every SDK.
  │ runner │ ────────────────────────────────► │ Flagsmith │
  │        │                                   └───────────┘
  │        │  stdio  ┌──────────────────────────────┐  ▲
- │        │ ──────► │ harness ─► SDK steps file    │──┘ SDK API calls
- └────────┘         │ (SDK image, built from source)│
-                    └──────────────────────────────┘
+ │        │ ──────► │ harness ─► SDK steps file    │──┤ SDK API calls
+ │        │         │ (SDK image, built from source)│  │
+ │        │         └──────────────────────────────┘  │
+ │        │  read what the SDK sent  ┌────────────┐   │
+ └────────┘ ───────────────────────► │ events API │ ◄─┘
+                                     └────────────┘
 ```
 
 Each scenario gets its own Flagsmith project and environment, and the steps split three ways:
@@ -36,6 +40,11 @@ Each scenario gets its own Flagsmith project and environment, and the steps spli
 The feature files are therefore the whole contract: the `When` steps in `features/` are the steps
 SDKs implement. The runner creates the scenario's SDK client just before its first `When` step.
 
+SDKs send tracking and exposure events to Flagsmith's events API, which the open-source Flagsmith
+image does not serve. [`events-api/`](events-api) stands in for it: scenarios that say
+`Given the SDK sends events to Flagsmith` get a client pointed at it with events enabled, and the
+runner reads back what the SDK sent.
+
 ## Writing scenarios
 
 - Describe behaviour, not wire details: assert on what the application receives and what Flagsmith
@@ -81,6 +90,9 @@ Each release also pins the Flagsmith version in `compose.yml` that it was verifi
 
    - `Create the SDK client` receives a server-side environment key and the Flagsmith API URL, and
      returns a client created with default options.
+   - `Create the SDK client with events enabled` receives the same two, then the events API URL,
+     and returns a client that sends events there. Only `@events` scenarios use it, so an SDK
+     without event tracking can leave it out.
    - A `When` step receives the client, then the step's placeholders in order, then its data table.
      A `<key> | value` table arrives as a map, e.g. `{"plan": "premium", "age": 30}`.
    - Values are JSON values. Pass flag and trait values through with their types intact.
@@ -155,6 +167,11 @@ uv run pytest -v
 | `FLAGSMITH_SDK_URL` | `$FLAGSMITH_URL`        | Where the SDK reaches Flagsmith                 |
 | `FLAGSMITH_PORT`    | `8000`                  | Host port for Flagsmith (`compose.yml`)         |
 | `FLAGSMITH_IMAGE`   | pinned version          | Override the Flagsmith image (`compose.yml`)    |
+| `EVENTS_URL`        | `http://localhost:8001` | Where the runner reaches the events API         |
+| `EVENTS_SDK_URL`    | `http://events:8000`    | Where the SDK reaches the events API            |
+| `EVENTS_PORT`       | `8001`                  | Host port for the events API (`compose.yml`)    |
 
 `SDK_COMMAND` can be any command, so you can also run a harness straight against an SDK checkout,
 without Docker, e.g. `python ../harness/python/harness.py ../../flagsmith-python-client/sdk-spec/steps.py`.
+An SDK running on the host reaches the two services on the host's ports, so set
+`FLAGSMITH_SDK_URL=$FLAGSMITH_URL` and `EVENTS_SDK_URL=$EVENTS_URL` with it.
diff --git a/compose.yml b/compose.yml
index d72d1a2..5b5bdcd 100644
--- a/compose.yml
+++ b/compose.yml
@@ -34,3 +34,18 @@ services:
       postgres:
         condition: service_healthy
 
+  events:
+    # Flagsmith's events API, which the open-source image does not serve, recording what
+    # SDKs send so the runner can check it (see events-api/events_api.py).
+    image: docker.io/library/python:3.13-alpine
+    command: python /app/events_api.py
+    volumes:
+      - ./events-api/events_api.py:/app/events_api.py:ro
+    ports:
+      - ${EVENTS_PORT:-8001}:8000
+    healthcheck:
+      test: ['CMD', 'python', '-c', 'import urllib.request; urllib.request.urlopen("http://localhost:8000/health")']
+      interval: 2s
+      timeout: 2s
+      retries: 30
+
diff --git a/events-api/events_api.py b/events-api/events_api.py
new file mode 100644
index 0000000..0fcf0e8
--- /dev/null
+++ b/events-api/events_api.py
@@ -0,0 +1,77 @@
+"""Flagsmith's events API, as far as the spec is concerned. See ../README.md.
+
+SDKs send tracking and exposure events to Flagsmith's events API, which the open-source
+Flagsmith image does not serve. The spec runs this in its place: it records every batch it
+is sent, keyed by the environment key that sent it, and the runner reads them back to
+check what an SDK sent.
+
+Uses only the standard library, so it runs in a stock Python image.
+"""
+
+import json
+import os
+import threading
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from typing import Any
+from urllib.parse import parse_qs, urlparse
+
+PORT = int(os.environ.get("PORT", 8000))
+MAX_BODY_BYTES = 10 * 1024 * 1024
+
+_lock = threading.Lock()
+_events: dict[str, list[dict[str, Any]]] = {}
+
+
+def record(environment_key: str, events: list[dict[str, Any]]) -> None:
+    with _lock:
+        _events.setdefault(environment_key, []).extend(events)
+
+
+def recorded(environment_key: str) -> list[dict[str, Any]]:
+    with _lock:
+        return list(_events.get(environment_key, ()))
+
+
+class Handler(BaseHTTPRequestHandler):
+    protocol_version = "HTTP/1.1"
+
+    def do_GET(self) -> None:  # noqa: N802 - BaseHTTPRequestHandler's naming
+        url = urlparse(self.path)
+        if url.path == "/health":
+            self.respond(200, {"status": "ok"})
+        elif url.path == "/spec/events":
+            key = parse_qs(url.query).get("environment_key", [""])[0]
+            self.respond(200, {"events": recorded(key)})
+        else:
+            self.respond(404, {"detail": f"Nothing is served at {url.path}"})
+
+    def do_POST(self) -> None:  # noqa: N802 - BaseHTTPRequestHandler's naming
+        # Any path: which one an SDK posts a batch to is a wire detail the spec leaves open.
+        environment_key = self.headers.get("X-Environment-Key")
+        if not environment_key:
+            self.respond(401, {"detail": "X-Environment-Key is missing"})
+            return
+        length = int(self.headers.get("Content-Length") or 0)
+        if length > MAX_BODY_BYTES:
+            self.respond(413, {"detail": "The batch is too large"})
+            return
+        try:
+            batch = json.loads(self.rfile.read(length) or b"{}")
+            events = batch["events"]
+        except (ValueError, KeyError, TypeError):
+            self.respond(400, {"detail": 'The body must be {"events": [...]}'})
+            return
+        record(environment_key, events)
+        self.respond(200, {"success": True})
+
+    def respond(self, status: int, body: dict[str, Any]) -> None:
+        raw = json.dumps(body).encode()
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(raw)))
+        self.end_headers()
+        self.wfile.write(raw)
+
+
+if __name__ == "__main__":
+    ThreadingHTTPServer(("", PORT), Handler).serve_forever()
diff --git a/features/get_experiment_flag.feature b/features/get_experiment_flag.feature
new file mode 100644
index 0000000..8e4fd78
--- /dev/null
+++ b/features/get_experiment_flag.feature
@@ -0,0 +1,38 @@
+@events
+Feature: Get an experiment flag
+  An SDK resolves a feature for an identity and records the exposure, in one call, so that
+  an application can show a variant and have it measured. Only an identity enrolled in a
+  running experiment is exposed: every other evaluation is left out of the data, so that
+  an experiment's results count the identities it actually reached.
+
+  Background:
+    Given a Flagsmith environment
+    And the SDK sends events to Flagsmith
+
+  Scenario: An experiment flag is resolved for the identity, with its traits
+    Given the environment has the following features:
+      | feature | enabled | value   |
+      | banner  | true    | "hello" |
+    And a segment "premium_users" where trait "plan" equals "premium"
+    And segment "premium_users" overrides feature "banner" with enabled true and value "hello premium user"
+    When the application gets the experiment flag "banner" for identity "user_123" with traits:
+      | trait | value     |
+      | plan  | "premium" |
+    Then the application receives the following flags:
+      | feature | enabled | value                |
+      | banner  | true    | "hello premium user" |
+    And Flagsmith has identity "user_123" with the following traits:
+      | trait | value     |
+      | plan  | "premium" |
+
+  Scenario: An identity outside a running experiment is not exposed
+    Given the environment has the following features:
+      | feature | enabled | value   |
+      | banner  | true    | "hello" |
+    When the application gets the experiment flag "banner" for identity "user_123" with traits:
+      | trait | value     |
+      | plan  | "premium" |
+    Then the application receives the following flags:
+      | feature | enabled | value   |
+      | banner  | true    | "hello" |
+    And Flagsmith has received no events
diff --git a/features/track_event.feature b/features/track_event.feature
new file mode 100644
index 0000000..aa298b7
--- /dev/null
+++ b/features/track_event.feature
@@ -0,0 +1,17 @@
+@events
+Feature: Track an event
+  An SDK sends an application's tracking events to Flagsmith, so that an experiment can
+  measure what the identities it exposed went on to do. Values reach Flagsmith as text,
+  whatever type the application tracked them as.
+
+  Background:
+    Given a Flagsmith environment
+    And the SDK sends events to Flagsmith
+
+  Scenario: A tracked event reaches Flagsmith
+    When the application tracks event "checkout_completed" for identity "user_123" with value 42 and metadata:
+      | key      | value |
+      | currency | "GBP" |
+    Then Flagsmith has received the following events:
+      | event              | identifier | value | metadata            |
+      | checkout_completed | user_123   | "42"  | {"currency": "GBP"} |
diff --git a/features/track_exposure_event.feature b/features/track_exposure_event.feature
new file mode 100644
index 0000000..e09465f
--- /dev/null
+++ b/features/track_exposure_event.feature
@@ -0,0 +1,16 @@
+@events
+Feature: Track an exposure event
+  An SDK records which variant of a feature an identity was shown, as the reserved
+  "$flag_exposure" event, so that an experiment can attribute later events to it.
+
+  Background:
+    Given a Flagsmith environment
+    And the SDK sends events to Flagsmith
+
+  Scenario: An exposure reaches Flagsmith as a $flag_exposure event
+    When the application tracks an exposure of feature "banner" for identity "user_123" with value "control" and metadata:
+      | key           | value |
+      | experiment_id | 1     |
+    Then Flagsmith has received the following events:
+      | event          | feature | identifier | value     | metadata             |
+      | $flag_exposure | banner  | user_123   | "control" | {"experiment_id": 1} |
diff --git a/harness/dart/README.md b/harness/dart/README.md
index b094c4b..e2151f6 100644
--- a/harness/dart/README.md
+++ b/harness/dart/README.md
@@ -12,7 +12,7 @@ step text. At runtime, `harness.dart` calls them with `Function.apply`.
 
 - Step functions are public, top-level functions in `sdk-spec/steps.dart`, each below its step marker
   (`// Create the SDK client`, `// When …`). Import the SDK by its package name.
-- `Create the SDK client` returns the client. `When` steps take it first.
+- A `Create the SDK client…` step returns the client. `When` steps take it first.
 - Arguments arrive as decoded JSON: `String`, `int`, `double`, `bool`, `null`, and data tables as
   `Map<String, dynamic>`. Declare them with those types, e.g. `Map<String, Object?> traits`.
 - Return a JSON-encodable value, or a `Future` of one.
diff --git a/harness/dart/build.dart b/harness/dart/build.dart
index 7895766..a9e64db 100644
--- a/harness/dart/build.dart
+++ b/harness/dart/build.dart
@@ -8,7 +8,7 @@
 import 'dart:convert';
 import 'dart:io';
 
-final _marker = RegExp(r'^\s*//+\s*(When .+?|Create the SDK client)\s*$');
+final _marker = RegExp(r'^\s*//+\s*(When .+?|Create the SDK client.*?)\s*$');
 // A top-level function declaration: an optional return type, then the name and its parameters.
 final _function = RegExp(
   r'^(?:[\w$][\w$<>?,.\[\] ]*\s)?([A-Za-z_$][\w$]*)\s*(?:<[^()]*>)?\s*\(',
diff --git a/harness/dotnet/Harness.cs b/harness/dotnet/Harness.cs
index b5beaa3..1565247 100644
--- a/harness/dotnet/Harness.cs
+++ b/harness/dotnet/Harness.cs
@@ -15,7 +15,7 @@ namespace SdkSpec.Harness;
 
 static partial class Program
 {
-    [GeneratedRegex(@"^\s*//+\s*(When .+?|Create the SDK client)\s*$")]
+    [GeneratedRegex(@"^\s*//+\s*(When .+?|Create the SDK client.*?)\s*$")]
     private static partial Regex Marker();
 
     // A static method declaration: modifiers including `static`, a return type, then the name.
diff --git a/harness/elixir/harness.exs b/harness/elixir/harness.exs
index 51c037f..bb5674d 100644
--- a/harness/elixir/harness.exs
+++ b/harness/elixir/harness.exs
@@ -5,7 +5,7 @@
 # Requires Jason, which the SDK is expected to depend on.
 
 defmodule SDKSpec.Harness do
-  @marker ~r/^\s*#+\s*(When .+?|Create the SDK client)\s*$/
+  @marker ~r/^\s*#+\s*(When .+?|Create the SDK client.*?)\s*$/
   @function ~r/^\s*def\s+([a-z_][a-zA-Z0-9_]*[?!]?)/
 
   def main([path]) do
diff --git a/harness/go/build.go b/harness/go/build.go
index 031817a..8815a89 100644
--- a/harness/go/build.go
+++ b/harness/go/build.go
@@ -26,7 +26,7 @@ import (
 )
 
 var (
-	marker = regexp.MustCompile(`^//\s*(When .+?|Create the SDK client)\s*$`)
+	marker = regexp.MustCompile(`^//\s*(When .+?|Create the SDK client.*?)\s*$`)
 	module = regexp.MustCompile(`(?m)^module\s+(\S+)`)
 )
 
diff --git a/harness/java/Harness.java b/harness/java/Harness.java
index b20db81..f0e918e 100644
--- a/harness/java/Harness.java
+++ b/harness/java/Harness.java
@@ -31,7 +31,7 @@
 public class Harness {
 
   private static final Pattern MARKER =
-      Pattern.compile("^\\s*//+\\s*(When .+?|Create the SDK client)\\s*$");
+      Pattern.compile("^\\s*//+\\s*(When .+?|Create the SDK client.*?)\\s*$");
   private static final Pattern METHOD =
       Pattern.compile("^\\s*(?:[\\w@]+\\s+)*static\\s+[^=;(]+?\\s(\\w+)\\s*\\(");
   private static final Pattern PACKAGE = Pattern.compile("^\\s*package\\s+([\\w.]+)\\s*;");
diff --git a/harness/kotlin/Harness.kt b/harness/kotlin/Harness.kt
index 6bb1559..7562775 100644
--- a/harness/kotlin/Harness.kt
+++ b/harness/kotlin/Harness.kt
@@ -19,7 +19,7 @@ import java.lang.reflect.Method
 import java.lang.reflect.Modifier
 import kotlin.system.exitProcess
 
-private val MARKER = Regex("""^\s*//+\s*(When .+?|Create the SDK client)\s*$""")
+private val MARKER = Regex("""^\s*//+\s*(When .+?|Create the SDK client.*?)\s*$""")
 private val FUNCTION = Regex("""^\s*(?:[\w@]+\s+)*fun\s+(\w+)\s*\(""")
 private val PACKAGE = Regex("""^\s*package\s+([\w.]+)""")
 private val JVM_NAME = Regex("""^\s*@file:JvmName\("(\w+)"\)""")
diff --git a/harness/nodejs/harness.mts b/harness/nodejs/harness.mts
index b815499..2be2d95 100644
--- a/harness/nodejs/harness.mts
+++ b/harness/nodejs/harness.mts
@@ -11,7 +11,7 @@ import { pathToFileURL } from 'node:url';
 
 type Step = (...args: unknown[]) => unknown;
 
-const MARKER = /^\s*\/\/+\s*(When .+?|Create the SDK client)\s*$/;
+const MARKER = /^\s*\/\/+\s*(When .+?|Create the SDK client.*?)\s*$/;
 const FUNCTION =
     /^\s*(?:export\s+)?(?:async\s+)?function\s*\*?\s*(\w+)|^\s*(?:export\s+)?(?:const|let|var)\s+(\w+)\s*=/;
 
diff --git a/harness/php/harness.php b/harness/php/harness.php
index b4550a2..5c0b17e 100644
--- a/harness/php/harness.php
+++ b/harness/php/harness.php
@@ -12,7 +12,7 @@
 
 namespace FlagsmithSdkSpec\Harness;
 
-const MARKER = '/^\s*(?:\/\/|#)\s*(When .+?|Create the SDK client)\s*$/';
+const MARKER = '/^\s*(?:\/\/|#)\s*(When .+?|Create the SDK client.*?)\s*$/';
 const FUNCTION_ = '/^\s*function\s+(\w+)\s*\(/';
 const NAMESPACE_ = '/^\s*namespace\s+([\w\\\\]+)\s*;/';
 
diff --git a/harness/python/harness.py b/harness/python/harness.py
index 6de22cc..c603de1 100644
--- a/harness/python/harness.py
+++ b/harness/python/harness.py
@@ -12,7 +12,7 @@
 from pathlib import Path
 from typing import Any, Callable
 
-MARKER = re.compile(r"^\s*#+\s*(When .+?|Create the SDK client)\s*$")
+MARKER = re.compile(r"^\s*#+\s*(When .+?|Create the SDK client.*?)\s*$")
 FUNCTION = re.compile(r"^\s*(?:async\s+)?def\s+(\w+)")
 
 
diff --git a/harness/ruby/harness.rb b/harness/ruby/harness.rb
index ce92ab3..4103390 100644
--- a/harness/ruby/harness.rb
+++ b/harness/ruby/harness.rb
@@ -6,7 +6,7 @@
 
 require 'json'
 
-MARKER = /\A\s*#+\s*(When .+?|Create the SDK client)\s*\z/
+MARKER = /\A\s*#+\s*(When .+?|Create the SDK client.*?)\s*\z/
 FUNCTION = /\A\s*def\s+(?:self\.)?(\w+[?!]?)/
 
 # Loads the steps file and maps each marked step's text to the method defined after it.
diff --git a/harness/rust/README.md b/harness/rust/README.md
index 326772c..cf64784 100644
--- a/harness/rust/README.md
+++ b/harness/rust/README.md
@@ -48,7 +48,7 @@ return value. [`src/main.rs`](src/main.rs) speaks the protocol over stdin and st
 ## Step functions
 
 - Step functions are `pub fn`, marked with a `//` comment holding the step's text.
-- `When` steps take the client first, as `&Client` or `&mut Client`. The `Create the SDK client`
+- `When` steps take the client first, as `&Client` or `&mut Client`. A `Create the SDK client…`
   function takes no client and returns one.
 - Other parameters take any type that deserialises from the JSON argument, e.g. `String` or
   `HashMap<String, V>` for a data table.
diff --git a/harness/rust/build.rs b/harness/rust/build.rs
index 2eb5ea6..fba402d 100644
--- a/harness/rust/build.rs
+++ b/harness/rust/build.rs
@@ -36,14 +36,14 @@ fn main() {
     fs::write(out, generate(&steps)).unwrap();
 }
 
-/// A step marker is a line comment holding `Create the SDK client` or a step starting with `When `.
+/// A step marker is a line comment starting with `Create the SDK client` or `When `.
 fn marker(line: &str) -> Option<&str> {
     let text = line
         .trim()
         .strip_prefix("//")?
         .trim_start_matches('/')
         .trim();
-    (text == "Create the SDK client" || text.starts_with("When ")).then_some(text)
+    (text.starts_with("Create the SDK client") || text.starts_with("When ")).then_some(text)
 }
 
 /// The name of the function defined on this line, e.g. `pub fn create_client(`.
@@ -150,7 +150,7 @@ fn generate(steps: &[Step]) -> String {
         if step.fallible {
             returned = format!("{returned}.map_err(error)?");
         }
-        // A step that does not use a client creates one: `Create the SDK client`.
+        // A step that does not use a client creates one: `Create the SDK client…`.
         let output = if step.client.is_some() {
             "returns_value"
         } else {
diff --git a/harness/swift/README.md b/harness/swift/README.md
index da0eeeb..11b75e3 100644
--- a/harness/swift/README.md
+++ b/harness/swift/README.md
@@ -66,7 +66,7 @@ result.
 
 - Step functions are `public`, top-level functions, each directly below its step marker comment
   (`// When …`). Give each a unique name: the table refers to them by name.
-- `When` steps take the client first. The `Create the SDK client` function takes no client and
+- `When` steps take the client first. A `Create the SDK client…` function takes no client and
   returns one. The client can be any type.
 - Other parameters take any `Decodable` type, e.g. `String` or `[String: V]` for a data table.
 - Return any `Encodable` type. Functions can be `async` and can `throw`. A thrown error is reported as
diff --git a/harness/swift/StepsGenerator/Sources/StepsTable/StepsTable.swift b/harness/swift/StepsGenerator/Sources/StepsTable/StepsTable.swift
index a80ea0f..a90008b 100644
--- a/harness/swift/StepsGenerator/Sources/StepsTable/StepsTable.swift
+++ b/harness/swift/StepsGenerator/Sources/StepsTable/StepsTable.swift
@@ -39,11 +39,11 @@ public func generateTable(_ steps: [Step], from stepsFile: String) -> String {
     return code + "]\n"
 }
 
-/// A step marker is a line comment holding `Create the SDK client` or a step starting with `When `.
+/// A step marker is a line comment starting with `Create the SDK client` or `When `.
 func marker(_ line: String) -> String? {
     guard line.hasPrefix("//") else { return nil }
     let text = line.drop { $0 == "/" }.trimmingCharacters(in: .whitespaces)
-    return text == "Create the SDK client" || text.hasPrefix("When ") ? text : nil
+    return text.hasPrefix("Create the SDK client") || text.hasPrefix("When ") ? text : nil
 }
 
 // `func` after any attributes and modifiers, e.g. `@MainActor public func createClient(`.
diff --git a/harness/swift/StepsGenerator/Tests/StepsTableTests/StepsTableTests.swift b/harness/swift/StepsGenerator/Tests/StepsTableTests/StepsTableTests.swift
index 046cab2..1f77678 100644
--- a/harness/swift/StepsGenerator/Tests/StepsTableTests/StepsTableTests.swift
+++ b/harness/swift/StepsGenerator/Tests/StepsTableTests/StepsTableTests.swift
@@ -23,6 +23,25 @@ final class StepsTableTests: XCTestCase {
         ])
     }
 
+    func testFindsEveryClientCreatingMarker() {
+        let source = """
+        // Create the SDK client
+        public func createClient(environmentKey: String, apiUrl: String) -> Client {
+            Client(environmentKey: environmentKey, apiUrl: apiUrl)
+        }
+
+        // Create the SDK client with events enabled
+        public func createClientWithEvents(environmentKey: String, apiUrl: String, eventsApiUrl: String) -> Client {
+            Client(environmentKey: environmentKey, apiUrl: apiUrl, eventsApiUrl: eventsApiUrl)
+        }
+        """
+
+        XCTAssertEqual(findSteps(in: source), [
+            Step(text: "Create the SDK client", function: "createClient"),
+            Step(text: "Create the SDK client with events enabled", function: "createClientWithEvents"),
+        ])
+    }
+
     func testIgnoresFuncInDocumentationComments() {
         let source = """
         // When the application gets the environment flags
diff --git a/runner/pyproject.toml b/runner/pyproject.toml
index 0e8a8ea..4dc0f2a 100644
--- a/runner/pyproject.toml
+++ b/runner/pyproject.toml
@@ -17,4 +17,5 @@ testpaths = ["tests"]
 addopts = "--strict-markers -p no:cacheprovider"
 markers = [
     "core: Behaviour every SDK must implement",
+    "events: Tracking and exposure events, powering experimentation",
 ]
diff --git a/runner/sdk_spec/events_api.py b/runner/sdk_spec/events_api.py
new file mode 100644
index 0000000..b3e1a15
--- /dev/null
+++ b/runner/sdk_spec/events_api.py
@@ -0,0 +1,40 @@
+"""Reads back the events an SDK sent. See events-api/events_api.py."""
+
+import time
+from typing import Any
+from urllib.parse import quote
+
+from sdk_spec.http import json_request, wait_until_ok
+
+# SDKs buffer events and send them on a timer, so the runner waits for them to arrive.
+ARRIVAL_TIMEOUT_SECONDS = 30
+# How long an SDK is given to send an event before the spec calls it unsent. Longer than
+# any SDK's send interval, so that a late batch fails the scenario rather than passing it.
+SETTLE_SECONDS = 15
+
+
+class EventsAPI:
+    def __init__(self, base_url: str) -> None:
+        self.base_url = base_url.rstrip("/")
+
+    def wait_until_healthy(self, timeout_seconds: float = 60) -> None:
+        wait_until_ok(f"{self.base_url}/health", timeout_seconds)
+
+    def events(self, environment_key: str) -> list[dict[str, Any]]:
+        """The events sent so far by the client holding `environment_key`."""
+        url = f"{self.base_url}/spec/events?environment_key={quote(environment_key)}"
+        return json_request("GET", url)["events"]
+
+    def wait_for_events(self, environment_key: str, count: int) -> list[dict[str, Any]]:
+        """Waits for `count` events to arrive, then returns everything that has."""
+        deadline = time.monotonic() + ARRIVAL_TIMEOUT_SECONDS
+        while True:
+            events = self.events(environment_key)
+            if len(events) >= count or time.monotonic() >= deadline:
+                return events
+            time.sleep(0.5)
+
+    def settle(self, environment_key: str) -> list[dict[str, Any]]:
+        """Waits out every SDK's send interval, then returns whatever arrived."""
+        time.sleep(SETTLE_SECONDS)
+        return self.events(environment_key)
diff --git a/runner/tests/conftest.py b/runner/tests/conftest.py
index b3ebc19..7562a99 100644
--- a/runner/tests/conftest.py
+++ b/runner/tests/conftest.py
@@ -9,6 +9,7 @@
 import pytest
 from pytest_bdd import given, parsers, then, when
 
+from sdk_spec.events_api import EventsAPI
 from sdk_spec.flagsmith_admin import Environment, FlagsmithAdmin
 from sdk_spec import sdk as sdk_module
 from sdk_spec.sdk import SDK
@@ -18,6 +19,11 @@
 # Where the SDK under test reaches Flagsmith. Differs from FLAGSMITH_URL when the
 # SDK runs on a different network, e.g. a container network.
 FLAGSMITH_SDK_URL = os.environ.get("FLAGSMITH_SDK_URL", FLAGSMITH_URL)
+# Where the runner reaches the events API, to read back what an SDK sent.
+EVENTS_URL = os.environ.get("EVENTS_URL", "http://localhost:8001")
+# Where the SDK under test reaches the events API. Unlike Flagsmith, which can be anywhere,
+# the events API is part of the spec's compose project, so it defaults to its service there.
+EVENTS_SDK_URL = os.environ.get("EVENTS_SDK_URL", "http://events:8000")
 # Starts the SDK's language harness, e.g. `docker run -i --rm sdk-spec-sdk`.
 SDK_COMMAND = os.environ.get("SDK_COMMAND", "")
 
@@ -25,6 +31,7 @@
 def pytest_report_header() -> list[str]:
     return [
         f"flagsmith: {FLAGSMITH_URL} (SDK uses {FLAGSMITH_SDK_URL})",
+        f"events: {EVENTS_URL} (SDK uses {EVENTS_SDK_URL})",
         f"sdk: {SDK_COMMAND}",
         f"missing steps: {sdk_module.MISSING_STEPS}",
     ]
@@ -48,6 +55,7 @@ def pytest_terminal_summary(terminalreporter: pytest.TerminalReporter) -> None:
 class ScenarioState:
     environment: Environment | None = None
     client: str | None = None
+    events: bool = False
     result: Any = None
 
 
@@ -59,12 +67,19 @@ def flagsmith() -> FlagsmithAdmin:
     return admin
 
 
+@pytest.fixture(scope="session")
+def events_api() -> EventsAPI:
+    events = EventsAPI(EVENTS_URL)
+    events.wait_until_healthy()
+    return events
+
+
 @pytest.fixture(scope="session")
 def sdk(pytestconfig: pytest.Config) -> Iterator[SDK]:
     if not SDK_COMMAND:
         pytest.exit("Set SDK_COMMAND to the command that starts the SDK harness", returncode=4)
     sdk = SDK(SDK_COMMAND)
-    required = [CREATE_CLIENT, *SDK_STEPS]
+    required = [CREATE_CLIENT, CREATE_CLIENT_WITH_EVENTS, *SDK_STEPS]
     implemented = [step for step in required if step in sdk.implemented_steps]
     # Write through the terminal reporter, so the line is shown even with output captured.
     terminal = pytestconfig.pluginmanager.get_plugin("terminalreporter")
@@ -119,6 +134,9 @@ def require_environment(state: ScenarioState) -> Environment:
 # others are JSON literals. Register SDK steps with `sdk_step`, never with `@when`.
 
 CREATE_CLIENT = "Create the SDK client"
+# A second client, for scenarios that are about the events an SDK sends. Events are off by
+# default in every SDK, and the spec's events API stands in for Flagsmith's.
+CREATE_CLIENT_WITH_EVENTS = "Create the SDK client with events enabled"
 SDK_STEPS: list[str] = []
 PLACEHOLDER = re.compile(r'("?)\{(\w+)\}\1')
 
@@ -126,10 +144,14 @@ def require_environment(state: ScenarioState) -> Environment:
 def sdk_client(state: ScenarioState, sdk: SDK) -> str:
     """Creates the scenario's SDK client before its first SDK step, after all `Given` steps."""
     if state.client is None:
+        environment = require_environment(state)
+        # Server-side SDKs authenticate with a server-side environment key.
+        args = [environment.server_side_key, f"{FLAGSMITH_SDK_URL.rstrip('/')}/api/v1/"]
+        if state.events:
+            args.append(f"{EVENTS_SDK_URL.rstrip('/')}/")
         state.client = sdk.step(
-            CREATE_CLIENT,
-            # Server-side SDKs authenticate with a server-side environment key.
-            [require_environment(state).server_side_key, f"{FLAGSMITH_SDK_URL.rstrip('/')}/api/v1/"],
+            CREATE_CLIENT_WITH_EVENTS if state.events else CREATE_CLIENT,
+            args,
             returns_client=True,
         )
     return state.client
@@ -164,6 +186,18 @@ def run(**kwargs: Any) -> None:
 
 sdk_step("the application gets the environment flags")
 sdk_step('the application gets the flags for identity "{identifier}" with traits:')
+sdk_step(
+    'the application gets the experiment flag "{feature}" for identity "{identifier}"'
+    " with traits:"
+)
+sdk_step(
+    'the application tracks event "{event}" for identity "{identifier}"'
+    " with value {value} and metadata:"
+)
+sdk_step(
+    'the application tracks an exposure of feature "{feature}" for identity "{identifier}"'
+    " with value {value} and metadata:"
+)
 
 
 # Given: the runner sets up Flagsmith
@@ -174,6 +208,13 @@ def flagsmith_environment(state: ScenarioState, flagsmith: FlagsmithAdmin) -> No
     state.environment = flagsmith.create_environment()
 
 
+@given("the SDK sends events to Flagsmith")
+def sdk_sends_events(state: ScenarioState) -> None:
+    """Asks for a client with events enabled, which the runner creates for the first SDK step."""
+    assert state.client is None, "The SDK client has already been created without events"
+    state.events = True
+
+
 @given("the environment has the following features:")
 def environment_features(
     state: ScenarioState, flagsmith: FlagsmithAdmin, datatable: list[list[str]]
@@ -241,3 +282,42 @@ def persisted_traits(
 ) -> None:
     actual = flagsmith.get_identity_traits(require_environment(state), identifier)
     assert actual == table_traits(datatable)
+
+
+# Columns of an event table that name the field differently from the event itself.
+EVENT_FIELDS = {"feature": "feature_name"}
+# Columns of an event table holding a JSON value rather than a bare name.
+EVENT_VALUES = ("value", "traits", "metadata")
+
+
+def event_as_expected(event: dict[str, Any], expectation: dict[str, Any]) -> dict[str, Any]:
+    """The part of a received event its expectation names, in the expectation's terms."""
+    actual = {column: event.get(EVENT_FIELDS.get(column, column)) for column in expectation}
+    if "metadata" in expectation:
+        # SDKs add metadata of their own, such as the version they report themselves as.
+        received = actual["metadata"] or {}
+        actual["metadata"] = {key: received.get(key) for key in expectation["metadata"]}
+    return actual
+
+
+@then("Flagsmith has received the following events:")
+def received_events(
+    state: ScenarioState, events_api: EventsAPI, datatable: list[list[str]]
+) -> None:
+    """Checks the events the SDK sent, on the columns the table names."""
+    expected = [
+        {key: parse_value(value) if key in EVENT_VALUES else value for key, value in row.items()}
+        for row in table_rows(datatable)
+    ]
+    received = events_api.wait_for_events(
+        require_environment(state).server_side_key, len(expected)
+    )
+    assert len(received) == len(expected), f"Flagsmith received {received}"
+    for event, expectation in zip(received, expected):
+        assert event_as_expected(event, expectation) == expectation, f"Flagsmith received {event}"
+
+
+@then("Flagsmith has received no events")
+def received_no_events(state: ScenarioState, events_api: EventsAPI) -> None:
+    received = events_api.settle(require_environment(state).server_side_key)
+    assert received == [], f"Flagsmith received {received}"

From 3342bac94b56bb8948a36af95a7ef4686bf32bab Mon Sep 17 00:00:00 2001
From: openhands <openhands@all-hands.dev>
Date: Wed, 21 Oct 2026 18:41:57 +0000
Subject: [PATCH 2/4] feat: Run @events scenarios against Flagsmith's live
 events pipeline

Replaces the events API stand-in with the services Flagsmith ships, mirroring
its own docker-compose.yml: the SDK posts to `flagsmith/events-ingestion-api`,
which queues to Kafka, from where `flagsmith/warehouse-delivery` inserts into
the environment's warehouse. The spec provisions a ClickHouse as that
warehouse and the runner reads the delivered events out of it.

`Given the SDK sends events to Flagsmith` now also connects the warehouse to
the scenario's environment, which is what makes the events API accept it.
Events are keyed by the environment's API key, as Flagsmith keys them.

Pins Flagsmith 2.287.0, the first release shipping the ClickHouse warehouse
and the db views the ingestion server reads, and turns experimentation on.

Also trims the three feature descriptions to a sentence, as reviewed.
---
 README.md                             |  44 ++++++----
 compose.yml                           | 118 +++++++++++++++++++++++---
 events-api/events_api.py              |  77 -----------------
 features/get_experiment_flag.feature  |   5 +-
 features/track_event.feature          |   4 +-
 features/track_exposure_event.feature |   3 +-
 runner/sdk_spec/events_api.py         |  40 ---------
 runner/sdk_spec/flagsmith_admin.py    |  12 +++
 runner/sdk_spec/http.py               |   7 +-
 runner/sdk_spec/warehouse.py          |  86 +++++++++++++++++++
 runner/tests/conftest.py              |  70 +++++++++------
 11 files changed, 284 insertions(+), 182 deletions(-)
 delete mode 100644 events-api/events_api.py
 delete mode 100644 runner/sdk_spec/events_api.py
 create mode 100644 runner/sdk_spec/warehouse.py

diff --git a/README.md b/README.md
index fa667d5..1b0d442 100644
--- a/README.md
+++ b/README.md
@@ -6,8 +6,7 @@ executed against every SDK.
 - [`features/`](features) — the spec. One `.feature` file per behaviour.
 - [`runner/`](runner) — executes the spec (pytest-bdd). Step definitions live here, written once for
   every SDK.
-- [`compose.yml`](compose.yml) — the Flagsmith API (pinned) the spec runs against.
-- [`events-api/`](events-api) — Flagsmith's events API, as far as the spec is concerned.
+- [`compose.yml`](compose.yml) — the Flagsmith services (pinned) the spec runs against.
 - [`harness/`](harness) — one harness per language, running an SDK's steps file
   ([`HARNESS.md`](HARNESS.md)).
 - [`.github/workflows/run-spec.yml`](.github/workflows/run-spec.yml) — reusable workflow SDKs call.
@@ -23,10 +22,14 @@ executed against every SDK.
  │        │  stdio  ┌──────────────────────────────┐  ▲
  │        │ ──────► │ harness ─► SDK steps file    │──┤ SDK API calls
  │        │         │ (SDK image, built from source)│  │
- │        │         └──────────────────────────────┘  │
- │        │  read what the SDK sent  ┌────────────┐   │
- └────────┘ ───────────────────────► │ events API │ ◄─┘
-                                     └────────────┘
+ │        │         └──────────────────────────────┘  │ events
+ │        │                                   ┌───────▼────┐
+ │        │                                   │ events API │
+ │        │                                   └───────┬────┘
+ │        │  read the delivered events                │ warehouse delivery
+ └────────┘ ────────────────────────────────► ┌───────▼───┐
+                                              │ warehouse │
+                                              └───────────┘
 ```
 
 Each scenario gets its own Flagsmith project and environment, and the steps split three ways:
@@ -40,10 +43,13 @@ Each scenario gets its own Flagsmith project and environment, and the steps spli
 The feature files are therefore the whole contract: the `When` steps in `features/` are the steps
 SDKs implement. The runner creates the scenario's SDK client just before its first `When` step.
 
-SDKs send tracking and exposure events to Flagsmith's events API, which the open-source Flagsmith
-image does not serve. [`events-api/`](events-api) stands in for it: scenarios that say
-`Given the SDK sends events to Flagsmith` get a client pointed at it with events enabled, and the
-runner reads back what the SDK sent.
+Tracking and exposure events take a longer path: an SDK sends them to Flagsmith's events API, which
+queues them for warehouse delivery to insert into the environment's warehouse.
+[`compose.yml`](compose.yml) runs that whole path, as Flagsmith's own `docker-compose.yml` does, with
+a ClickHouse standing in for the environment's warehouse. A scenario that says
+`Given the SDK sends events to Flagsmith` gets the warehouse connected to its environment — the
+events API only accepts an environment that has one — and a client with events enabled pointed at
+the events API. The runner then reads the delivered events out of the warehouse.
 
 ## Writing scenarios
 
@@ -90,9 +96,9 @@ Each release also pins the Flagsmith version in `compose.yml` that it was verifi
 
    - `Create the SDK client` receives a server-side environment key and the Flagsmith API URL, and
      returns a client created with default options.
-   - `Create the SDK client with events enabled` receives the same two, then the events API URL,
-     and returns a client that sends events there. Only `@events` scenarios use it, so an SDK
-     without event tracking can leave it out.
+   - `Create the SDK client with events enabled` receives the same two, then the base URL of
+     Flagsmith's events API, and returns a client that sends events there. Only `@events`
+     scenarios use it, so an SDK without event tracking can leave it out.
    - A `When` step receives the client, then the step's placeholders in order, then its data table.
      A `<key> | value` table arrives as a map, e.g. `{"plan": "premium", "age": 30}`.
    - Values are JSON values. Pass flag and trait values through with their types intact.
@@ -167,11 +173,13 @@ uv run pytest -v
 | `FLAGSMITH_SDK_URL` | `$FLAGSMITH_URL`        | Where the SDK reaches Flagsmith                 |
 | `FLAGSMITH_PORT`    | `8000`                  | Host port for Flagsmith (`compose.yml`)         |
 | `FLAGSMITH_IMAGE`   | pinned version          | Override the Flagsmith image (`compose.yml`)    |
-| `EVENTS_URL`        | `http://localhost:8001` | Where the runner reaches the events API         |
-| `EVENTS_SDK_URL`    | `http://events:8000`    | Where the SDK reaches the events API            |
-| `EVENTS_PORT`       | `8001`                  | Host port for the events API (`compose.yml`)    |
+| `EVENTS_SDK_URL`    | `http://events:8080`    | Where the SDK reaches the events API            |
+| `EVENTS_PORT`       | `8080`                  | Host port for the events API (`compose.yml`)    |
+| `WAREHOUSE_URL`     | `http://localhost:8123` | Where the runner reaches the warehouse          |
+| `WAREHOUSE_HOST`    | `clickhouse`            | Where Flagsmith reaches the warehouse           |
+| `CLICKHOUSE_PORT`   | `8123`                  | Host port for the warehouse (`compose.yml`)     |
 
 `SDK_COMMAND` can be any command, so you can also run a harness straight against an SDK checkout,
 without Docker, e.g. `python ../harness/python/harness.py ../../flagsmith-python-client/sdk-spec/steps.py`.
-An SDK running on the host reaches the two services on the host's ports, so set
-`FLAGSMITH_SDK_URL=$FLAGSMITH_URL` and `EVENTS_SDK_URL=$EVENTS_URL` with it.
+An SDK running on the host reaches the services on the host's ports, so set
+`FLAGSMITH_SDK_URL=$FLAGSMITH_URL` and `EVENTS_SDK_URL=http://localhost:8080` with it.
diff --git a/compose.yml b/compose.yml
index 5b5bdcd..ed57f42 100644
--- a/compose.yml
+++ b/compose.yml
@@ -1,7 +1,16 @@
 # The Flagsmith API the spec runs against. The runner runs on the host and starts the
 # SDK under test in a container on this compose project's network (see README.md).
+#
+# The experimentation services mirror Flagsmith's own docker-compose.yml, so that `@events`
+# scenarios run against the real ingestion and delivery path: an SDK posts events to the
+# ingestion server, which writes them to Kafka, from where warehouse delivery inserts them
+# into the environment's warehouse. The spec provisions ClickHouse as that warehouse and
+# reads the events back out of it.
 name: sdk-spec
 
+volumes:
+  clickhouse-data:
+
 services:
   postgres:
     image: postgres:15.5-alpine
@@ -16,13 +25,17 @@ services:
 
   flagsmith:
     # The Flagsmith version a spec release is verified against.
-    image: ${FLAGSMITH_IMAGE:-docker.io/flagsmith/flagsmith:2.276.0}
+    image: ${FLAGSMITH_IMAGE:-docker.io/flagsmith/flagsmith:2.287.0}
     command: migrate-and-serve
     environment:
       DATABASE_URL: postgresql://postgres:password@postgres:5432/flagsmith
       DJANGO_ALLOWED_HOSTS: '*'
       DJANGO_SECRET_KEY: sdk-spec
       ENVIRONMENT: test
+      EXPERIMENTATION_ENABLED: 'true'
+      # Encrypts the warehouse connection's credentials; warehouse delivery decrypts them
+      # with the same value.
+      WAREHOUSE_CREDENTIALS_SECRET: &warehouse-credentials-secret sdk-spec
     ports:
       - ${FLAGSMITH_PORT:-8000}:8000
     healthcheck:
@@ -34,18 +47,101 @@ services:
       postgres:
         condition: service_healthy
 
-  events:
-    # Flagsmith's events API, which the open-source image does not serve, recording what
-    # SDKs send so the runner can check it (see events-api/events_api.py).
-    image: docker.io/library/python:3.13-alpine
-    command: python /app/events_api.py
+  kafka:
+    image: apache/kafka:3.9.1
+    environment:
+      KAFKA_NODE_ID: 1
+      KAFKA_PROCESS_ROLES: broker,controller
+      KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
+      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
+      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
+      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
+    healthcheck:
+      test: ['CMD-SHELL', '/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092']
+      interval: 5s
+      timeout: 10s
+      retries: 30
+
+  kafka-init:
+    image: apache/kafka:3.9.1
+    entrypoint: ['/bin/sh', '-c']
+    command:
+      - |
+        for topic in flagsmith_warehouse_events external_warehouse_events external_warehouse_events_retry; do
+          /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka:9092 --create --if-not-exists --topic $$topic
+        done
+    depends_on:
+      kafka:
+        condition: service_healthy
+
+  # The environment's warehouse: where delivered events land and the runner reads them from.
+  clickhouse:
+    image: clickhouse/clickhouse-server:25.12
+    environment:
+      CLICKHOUSE_USER: flagsmith
+      CLICKHOUSE_PASSWORD: password
+    ulimits:
+      nofile:
+        soft: 262144
+        hard: 262144
     volumes:
-      - ./events-api/events_api.py:/app/events_api.py:ro
+      - clickhouse-data:/var/lib/clickhouse
     ports:
-      - ${EVENTS_PORT:-8001}:8000
+      - ${CLICKHOUSE_PORT:-8123}:8123
     healthcheck:
-      test: ['CMD', 'python', '-c', 'import urllib.request; urllib.request.urlopen("http://localhost:8000/health")']
-      interval: 2s
-      timeout: 2s
+      test: ['CMD', 'wget', '--spider', '-q', 'http://localhost:8123/ping']
+      interval: 3s
+      timeout: 3s
       retries: 30
 
+  # Creates the events table, from the schema the pinned Flagsmith version ships.
+  clickhouse-init:
+    image: clickhouse/clickhouse-server:25.12
+    entrypoint: ['/bin/sh', '-c']
+    command:
+      - |
+        set -e
+        wget -qO /tmp/events.sql https://raw.githubusercontent.com/Flagsmith/flagsmith/v2.287.0/experimentation/clickhouse/events.sql
+        clickhouse-client --host clickhouse --user flagsmith --password password --multiquery < /tmp/events.sql
+    depends_on:
+      clickhouse:
+        condition: service_healthy
+
+  # Flagsmith's events API: where SDKs send tracking and exposure events.
+  events:
+    image: flagsmith/events-ingestion-api:0.5.1
+    environment:
+      KAFKA_BOOTSTRAP_SERVERS: kafka:9092
+      KAFKA_AUTH: none
+      KAFKA_TOPIC: flagsmith_warehouse_events
+      KAFKA_EXTERNAL_TOPIC: external_warehouse_events
+      DATABASE_URL: postgres://postgres:password@postgres:5432/flagsmith
+    ports:
+      - ${EVENTS_PORT:-8080}:8080
+    depends_on:
+      flagsmith:
+        condition: service_healthy
+      kafka-init:
+        condition: service_completed_successfully
+
+  warehouse-delivery:
+    image: flagsmith/warehouse-delivery:0.4.0
+    environment:
+      KAFKA_BOOTSTRAP_SERVERS: kafka:9092
+      KAFKA_AUTH: none
+      DATABASE_URL: postgresql://postgres:password@postgres:5432/flagsmith
+      WAREHOUSE_CREDENTIALS_SECRET: *warehouse-credentials-secret
+      # The spec asserts on events as soon as they are sent, so deliver them promptly.
+      BATCH_MAX_WAIT_MS: 1000
+    depends_on:
+      flagsmith:
+        condition: service_healthy
+      kafka-init:
+        condition: service_completed_successfully
+      clickhouse-init:
+        condition: service_completed_successfully
diff --git a/events-api/events_api.py b/events-api/events_api.py
deleted file mode 100644
index 0fcf0e8..0000000
--- a/events-api/events_api.py
+++ /dev/null
@@ -1,77 +0,0 @@
-"""Flagsmith's events API, as far as the spec is concerned. See ../README.md.
-
-SDKs send tracking and exposure events to Flagsmith's events API, which the open-source
-Flagsmith image does not serve. The spec runs this in its place: it records every batch it
-is sent, keyed by the environment key that sent it, and the runner reads them back to
-check what an SDK sent.
-
-Uses only the standard library, so it runs in a stock Python image.
-"""
-
-import json
-import os
-import threading
-from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
-from typing import Any
-from urllib.parse import parse_qs, urlparse
-
-PORT = int(os.environ.get("PORT", 8000))
-MAX_BODY_BYTES = 10 * 1024 * 1024
-
-_lock = threading.Lock()
-_events: dict[str, list[dict[str, Any]]] = {}
-
-
-def record(environment_key: str, events: list[dict[str, Any]]) -> None:
-    with _lock:
-        _events.setdefault(environment_key, []).extend(events)
-
-
-def recorded(environment_key: str) -> list[dict[str, Any]]:
-    with _lock:
-        return list(_events.get(environment_key, ()))
-
-
-class Handler(BaseHTTPRequestHandler):
-    protocol_version = "HTTP/1.1"
-
-    def do_GET(self) -> None:  # noqa: N802 - BaseHTTPRequestHandler's naming
-        url = urlparse(self.path)
-        if url.path == "/health":
-            self.respond(200, {"status": "ok"})
-        elif url.path == "/spec/events":
-            key = parse_qs(url.query).get("environment_key", [""])[0]
-            self.respond(200, {"events": recorded(key)})
-        else:
-            self.respond(404, {"detail": f"Nothing is served at {url.path}"})
-
-    def do_POST(self) -> None:  # noqa: N802 - BaseHTTPRequestHandler's naming
-        # Any path: which one an SDK posts a batch to is a wire detail the spec leaves open.
-        environment_key = self.headers.get("X-Environment-Key")
-        if not environment_key:
-            self.respond(401, {"detail": "X-Environment-Key is missing"})
-            return
-        length = int(self.headers.get("Content-Length") or 0)
-        if length > MAX_BODY_BYTES:
-            self.respond(413, {"detail": "The batch is too large"})
-            return
-        try:
-            batch = json.loads(self.rfile.read(length) or b"{}")
-            events = batch["events"]
-        except (ValueError, KeyError, TypeError):
-            self.respond(400, {"detail": 'The body must be {"events": [...]}'})
-            return
-        record(environment_key, events)
-        self.respond(200, {"success": True})
-
-    def respond(self, status: int, body: dict[str, Any]) -> None:
-        raw = json.dumps(body).encode()
-        self.send_response(status)
-        self.send_header("Content-Type", "application/json")
-        self.send_header("Content-Length", str(len(raw)))
-        self.end_headers()
-        self.wfile.write(raw)
-
-
-if __name__ == "__main__":
-    ThreadingHTTPServer(("", PORT), Handler).serve_forever()
diff --git a/features/get_experiment_flag.feature b/features/get_experiment_flag.feature
index 8e4fd78..45c3eec 100644
--- a/features/get_experiment_flag.feature
+++ b/features/get_experiment_flag.feature
@@ -1,9 +1,6 @@
 @events
 Feature: Get an experiment flag
-  An SDK resolves a feature for an identity and records the exposure, in one call, so that
-  an application can show a variant and have it measured. Only an identity enrolled in a
-  running experiment is exposed: every other evaluation is left out of the data, so that
-  an experiment's results count the identities it actually reached.
+  An SDK resolves a feature for an identity and records the exposure.
 
   Background:
     Given a Flagsmith environment
diff --git a/features/track_event.feature b/features/track_event.feature
index aa298b7..aacf9c5 100644
--- a/features/track_event.feature
+++ b/features/track_event.feature
@@ -1,8 +1,6 @@
 @events
 Feature: Track an event
-  An SDK sends an application's tracking events to Flagsmith, so that an experiment can
-  measure what the identities it exposed went on to do. Values reach Flagsmith as text,
-  whatever type the application tracked them as.
+  An SDK sends an application's tracking events to Flagsmith.
 
   Background:
     Given a Flagsmith environment
diff --git a/features/track_exposure_event.feature b/features/track_exposure_event.feature
index e09465f..8c8ac86 100644
--- a/features/track_exposure_event.feature
+++ b/features/track_exposure_event.feature
@@ -1,7 +1,6 @@
 @events
 Feature: Track an exposure event
-  An SDK records which variant of a feature an identity was shown, as the reserved
-  "$flag_exposure" event, so that an experiment can attribute later events to it.
+  An SDK records which variant of a feature an identity was shown, as a "$flag_exposure" event.
 
   Background:
     Given a Flagsmith environment
diff --git a/runner/sdk_spec/events_api.py b/runner/sdk_spec/events_api.py
deleted file mode 100644
index b3e1a15..0000000
--- a/runner/sdk_spec/events_api.py
+++ /dev/null
@@ -1,40 +0,0 @@
-"""Reads back the events an SDK sent. See events-api/events_api.py."""
-
-import time
-from typing import Any
-from urllib.parse import quote
-
-from sdk_spec.http import json_request, wait_until_ok
-
-# SDKs buffer events and send them on a timer, so the runner waits for them to arrive.
-ARRIVAL_TIMEOUT_SECONDS = 30
-# How long an SDK is given to send an event before the spec calls it unsent. Longer than
-# any SDK's send interval, so that a late batch fails the scenario rather than passing it.
-SETTLE_SECONDS = 15
-
-
-class EventsAPI:
-    def __init__(self, base_url: str) -> None:
-        self.base_url = base_url.rstrip("/")
-
-    def wait_until_healthy(self, timeout_seconds: float = 60) -> None:
-        wait_until_ok(f"{self.base_url}/health", timeout_seconds)
-
-    def events(self, environment_key: str) -> list[dict[str, Any]]:
-        """The events sent so far by the client holding `environment_key`."""
-        url = f"{self.base_url}/spec/events?environment_key={quote(environment_key)}"
-        return json_request("GET", url)["events"]
-
-    def wait_for_events(self, environment_key: str, count: int) -> list[dict[str, Any]]:
-        """Waits for `count` events to arrive, then returns everything that has."""
-        deadline = time.monotonic() + ARRIVAL_TIMEOUT_SECONDS
-        while True:
-            events = self.events(environment_key)
-            if len(events) >= count or time.monotonic() >= deadline:
-                return events
-            time.sleep(0.5)
-
-    def settle(self, environment_key: str) -> list[dict[str, Any]]:
-        """Waits out every SDK's send interval, then returns whatever arrived."""
-        time.sleep(SETTLE_SECONDS)
-        return self.events(environment_key)
diff --git a/runner/sdk_spec/flagsmith_admin.py b/runner/sdk_spec/flagsmith_admin.py
index e0d9180..83ef35a 100644
--- a/runner/sdk_spec/flagsmith_admin.py
+++ b/runner/sdk_spec/flagsmith_admin.py
@@ -74,6 +74,18 @@ def create_environment(self) -> Environment:
     def delete_project(self, environment: Environment) -> None:
         self._request("DELETE", f"/projects/{environment.project_id}/")
 
+    def connect_warehouse(self, environment: Environment, config: dict[str, Any]) -> None:
+        """Connects the environment to a warehouse, so that it can receive events.
+
+        Flagsmith's events API only accepts an environment that has one, and warehouse
+        delivery inserts the environment's events there for the runner to read.
+        """
+        self._request(
+            "POST",
+            f"/environments/{environment.api_key}/warehouse-connections/",
+            {"name": "sdk-spec", "warehouse_type": "clickhouse", **config},
+        )
+
     def create_feature(
         self, environment: Environment, name: str, enabled: bool, value: Any
     ) -> None:
diff --git a/runner/sdk_spec/http.py b/runner/sdk_spec/http.py
index 9075498..85545dd 100644
--- a/runner/sdk_spec/http.py
+++ b/runner/sdk_spec/http.py
@@ -33,11 +33,14 @@ def json_request(
     return json.loads(raw) if raw else None
 
 
-def wait_until_ok(url: str, timeout_seconds: float) -> Any:
+def wait_until_ok(url: str, timeout_seconds: float, expect_json: bool = True) -> Any:
     deadline = time.monotonic() + timeout_seconds
     while True:
         try:
-            return json_request("GET", url, timeout=5)
+            if expect_json:
+                return json_request("GET", url, timeout=5)
+            with urllib.request.urlopen(url, timeout=5) as response:
+                return response.read()
         except (urllib.error.URLError, ConnectionError, HTTPError) as e:
             if time.monotonic() > deadline:
                 raise TimeoutError(f"{url} did not become healthy: {e}") from e
diff --git a/runner/sdk_spec/warehouse.py b/runner/sdk_spec/warehouse.py
new file mode 100644
index 0000000..0eb529f
--- /dev/null
+++ b/runner/sdk_spec/warehouse.py
@@ -0,0 +1,86 @@
+"""Reads the events an SDK sent out of the environment's warehouse.
+
+An SDK sends events to Flagsmith's events API, which writes them to Kafka, from where
+warehouse delivery inserts them into the environment's warehouse. The spec connects a
+ClickHouse warehouse to every `@events` environment (see `compose.yml`), so this reads the
+far end of the real delivery path rather than standing in for it.
+"""
+
+import json
+import time
+import urllib.request
+from typing import Any
+from urllib.parse import quote
+
+from sdk_spec.http import wait_until_ok
+
+# How long the path from an SDK's buffer to the warehouse is given. SDKs send on a timer of
+# their own, then the event is queued in Kafka and delivered in a batch.
+ARRIVAL_TIMEOUT_SECONDS = 60
+# How long an SDK is given to send an event before the spec calls it unsent. Longer than
+# any SDK's send interval, so that a late event fails the scenario rather than passing it.
+SETTLE_SECONDS = 30
+
+# The columns the spec asserts on. `traits` and `metadata` are stored as JSON text.
+COLUMNS = ("event", "feature_name", "identifier", "value", "traits", "metadata")
+JSON_COLUMNS = ("traits", "metadata")
+
+EVENTS_QUERY = f"""
+SELECT {", ".join(COLUMNS)}
+FROM events
+WHERE environment_key = {{environment_key:String}}
+ORDER BY timestamp, event
+"""
+
+
+class Warehouse:
+    """The ClickHouse instance the spec connects to its environments."""
+
+    def __init__(self, base_url: str, database: str, username: str, password: str) -> None:
+        self.base_url = base_url.rstrip("/")
+        self.database = database
+        self.username = username
+        self.password = password
+
+    def wait_until_healthy(self, timeout_seconds: float = 120) -> None:
+        wait_until_ok(f"{self.base_url}/ping", timeout_seconds, expect_json=False)
+
+    def query(self, sql: str, **parameters: str) -> list[dict[str, Any]]:
+        """Runs a query, with `{name:Type}` placeholders bound to `parameters`."""
+        url = f"{self.base_url}/?database={self.database}&default_format=JSONEachRow"
+        url += "".join(f"&param_{name}={quote(value)}" for name, value in parameters.items())
+        request = urllib.request.Request(
+            url,
+            data=sql.encode(),
+            method="POST",
+            headers={"X-ClickHouse-User": self.username, "X-ClickHouse-Key": self.password},
+        )
+        with urllib.request.urlopen(request, timeout=30) as response:
+            body = response.read().decode()
+        return [json.loads(line) for line in body.splitlines() if line]
+
+    def events(self, environment_key: str) -> list[dict[str, Any]]:
+        """The events delivered for the environment so far, oldest first.
+
+        An SDK sends events under its own environment key; Flagsmith records them against
+        the environment, so `environment_key` is the environment's API key.
+        """
+        rows = self.query(EVENTS_QUERY, environment_key=environment_key)
+        for row in rows:
+            for column in JSON_COLUMNS:
+                row[column] = json.loads(row[column]) if row[column] else None
+        return rows
+
+    def wait_for_events(self, environment_key: str, count: int) -> list[dict[str, Any]]:
+        """Waits for `count` events to be delivered, then returns everything that has."""
+        deadline = time.monotonic() + ARRIVAL_TIMEOUT_SECONDS
+        while True:
+            events = self.events(environment_key)
+            if len(events) >= count or time.monotonic() >= deadline:
+                return events
+            time.sleep(1)
+
+    def settle(self, environment_key: str) -> list[dict[str, Any]]:
+        """Waits out the send and delivery path, then returns whatever arrived."""
+        time.sleep(SETTLE_SECONDS)
+        return self.events(environment_key)
diff --git a/runner/tests/conftest.py b/runner/tests/conftest.py
index 7562a99..992220e 100644
--- a/runner/tests/conftest.py
+++ b/runner/tests/conftest.py
@@ -9,21 +9,32 @@
 import pytest
 from pytest_bdd import given, parsers, then, when
 
-from sdk_spec.events_api import EventsAPI
 from sdk_spec.flagsmith_admin import Environment, FlagsmithAdmin
 from sdk_spec import sdk as sdk_module
 from sdk_spec.sdk import SDK
+from sdk_spec.warehouse import Warehouse
 
 # Where the runner reaches Flagsmith, to seed and inspect it.
 FLAGSMITH_URL = os.environ.get("FLAGSMITH_URL", "http://localhost:8000")
 # Where the SDK under test reaches Flagsmith. Differs from FLAGSMITH_URL when the
 # SDK runs on a different network, e.g. a container network.
 FLAGSMITH_SDK_URL = os.environ.get("FLAGSMITH_SDK_URL", FLAGSMITH_URL)
-# Where the runner reaches the events API, to read back what an SDK sent.
-EVENTS_URL = os.environ.get("EVENTS_URL", "http://localhost:8001")
-# Where the SDK under test reaches the events API. Unlike Flagsmith, which can be anywhere,
-# the events API is part of the spec's compose project, so it defaults to its service there.
-EVENTS_SDK_URL = os.environ.get("EVENTS_SDK_URL", "http://events:8000")
+# Where the SDK under test reaches Flagsmith's events API. Unlike Flagsmith, which can be
+# anywhere, it is part of the spec's compose project, so it defaults to its service there.
+EVENTS_SDK_URL = os.environ.get("EVENTS_SDK_URL", "http://events:8080")
+# Where the runner reaches the warehouse the spec connects to its environments, to read
+# back the events delivered to it, and how Flagsmith and warehouse delivery reach it.
+WAREHOUSE_URL = os.environ.get("WAREHOUSE_URL", "http://localhost:8123")
+WAREHOUSE_CONFIG = {
+    "config": {
+        "host": os.environ.get("WAREHOUSE_HOST", "clickhouse"),
+        "port": 8123,
+        "database": "flagsmith_exp",
+        "username": "flagsmith",
+        "secure": False,
+    },
+    "credentials": {"password": "password"},
+}
 # Starts the SDK's language harness, e.g. `docker run -i --rm sdk-spec-sdk`.
 SDK_COMMAND = os.environ.get("SDK_COMMAND", "")
 
@@ -31,7 +42,7 @@
 def pytest_report_header() -> list[str]:
     return [
         f"flagsmith: {FLAGSMITH_URL} (SDK uses {FLAGSMITH_SDK_URL})",
-        f"events: {EVENTS_URL} (SDK uses {EVENTS_SDK_URL})",
+        f"events: {EVENTS_SDK_URL} (SDK), warehouse: {WAREHOUSE_URL} (runner)",
         f"sdk: {SDK_COMMAND}",
         f"missing steps: {sdk_module.MISSING_STEPS}",
     ]
@@ -68,10 +79,16 @@ def flagsmith() -> FlagsmithAdmin:
 
 
 @pytest.fixture(scope="session")
-def events_api() -> EventsAPI:
-    events = EventsAPI(EVENTS_URL)
-    events.wait_until_healthy()
-    return events
+def warehouse() -> Warehouse:
+    config = WAREHOUSE_CONFIG["config"]
+    warehouse = Warehouse(
+        WAREHOUSE_URL,
+        database=config["database"],
+        username=config["username"],
+        password=WAREHOUSE_CONFIG["credentials"]["password"],
+    )
+    warehouse.wait_until_healthy()
+    return warehouse
 
 
 @pytest.fixture(scope="session")
@@ -209,9 +226,14 @@ def flagsmith_environment(state: ScenarioState, flagsmith: FlagsmithAdmin) -> No
 
 
 @given("the SDK sends events to Flagsmith")
-def sdk_sends_events(state: ScenarioState) -> None:
-    """Asks for a client with events enabled, which the runner creates for the first SDK step."""
+def sdk_sends_events(state: ScenarioState, flagsmith: FlagsmithAdmin) -> None:
+    """Connects the environment's warehouse, so that Flagsmith accepts the SDK's events.
+
+    Also asks for a client with events enabled, which the runner creates for the first SDK
+    step: events are off by default in every SDK.
+    """
     assert state.client is None, "The SDK client has already been created without events"
+    flagsmith.connect_warehouse(require_environment(state), WAREHOUSE_CONFIG)
     state.events = True
 
 
@@ -291,33 +313,31 @@ def persisted_traits(
 
 
 def event_as_expected(event: dict[str, Any], expectation: dict[str, Any]) -> dict[str, Any]:
-    """The part of a received event its expectation names, in the expectation's terms."""
+    """The part of a delivered event its expectation names, in the expectation's terms."""
     actual = {column: event.get(EVENT_FIELDS.get(column, column)) for column in expectation}
     if "metadata" in expectation:
         # SDKs add metadata of their own, such as the version they report themselves as.
-        received = actual["metadata"] or {}
-        actual["metadata"] = {key: received.get(key) for key in expectation["metadata"]}
+        delivered = actual["metadata"] or {}
+        actual["metadata"] = {key: delivered.get(key) for key in expectation["metadata"]}
     return actual
 
 
 @then("Flagsmith has received the following events:")
 def received_events(
-    state: ScenarioState, events_api: EventsAPI, datatable: list[list[str]]
+    state: ScenarioState, warehouse: Warehouse, datatable: list[list[str]]
 ) -> None:
     """Checks the events the SDK sent, on the columns the table names."""
     expected = [
         {key: parse_value(value) if key in EVENT_VALUES else value for key, value in row.items()}
         for row in table_rows(datatable)
     ]
-    received = events_api.wait_for_events(
-        require_environment(state).server_side_key, len(expected)
-    )
-    assert len(received) == len(expected), f"Flagsmith received {received}"
-    for event, expectation in zip(received, expected):
+    delivered = warehouse.wait_for_events(require_environment(state).api_key, len(expected))
+    assert len(delivered) == len(expected), f"Flagsmith received {delivered}"
+    for event, expectation in zip(delivered, expected):
         assert event_as_expected(event, expectation) == expectation, f"Flagsmith received {event}"
 
 
 @then("Flagsmith has received no events")
-def received_no_events(state: ScenarioState, events_api: EventsAPI) -> None:
-    received = events_api.settle(require_environment(state).server_side_key)
-    assert received == [], f"Flagsmith received {received}"
+def received_no_events(state: ScenarioState, warehouse: Warehouse) -> None:
+    delivered = warehouse.settle(require_environment(state).api_key)
+    assert delivered == [], f"Flagsmith received {delivered}"

From 89db70c688a28df2c07bf4e5c25ff2c9ab734088 Mon Sep 17 00:00:00 2001
From: "flagsmithsmith[bot]"
 <346993984+flagsmithsmith[bot]@users.noreply.github.com>
Date: Wed, 28 Oct 2026 17:27:49 +0000
Subject: [PATCH 3/4] Fix a stale comment on the events client step

The spec no longer stands in for Flagsmith's events API.
---
 runner/tests/conftest.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/runner/tests/conftest.py b/runner/tests/conftest.py
index 992220e..1847d1c 100644
--- a/runner/tests/conftest.py
+++ b/runner/tests/conftest.py
@@ -152,7 +152,7 @@ def require_environment(state: ScenarioState) -> Environment:
 
 CREATE_CLIENT = "Create the SDK client"
 # A second client, for scenarios that are about the events an SDK sends. Events are off by
-# default in every SDK, and the spec's events API stands in for Flagsmith's.
+# default in every SDK, and the events API is not derived from the Flagsmith API URL.
 CREATE_CLIENT_WITH_EVENTS = "Create the SDK client with events enabled"
 SDK_STEPS: list[str] = []
 PLACEHOLDER = re.compile(r'("?)\{(\w+)\}\1')

From 0cd8cc93392f9c25507673f7341ca8fa2bf3b068 Mon Sep 17 00:00:00 2001
From: "flagsmithsmith[bot]"
 <346993984+flagsmithsmith[bot]@users.noreply.github.com>
Date: Wed, 28 Oct 2026 17:42:58 +0000
Subject: [PATCH 4/4] feat: Spec that getExperimentFlag records the exposure

The feature file said an SDK records the exposure, but only the negative
was covered: an identity outside a running experiment produces no event.
An SDK that resolves the flag and never records anything passed.

The runner now runs an experiment through the Admin API: a multivariate
option keyed by the variant, a rollout enrolling every identity and
allocating all of them to it, and the experiment started. One variant at
100% makes which variant an identity is served deterministic.
---
 features/get_experiment_flag.feature | 15 +++++++
 runner/sdk_spec/flagsmith_admin.py   | 65 ++++++++++++++++++++++++++--
 runner/tests/conftest.py             | 13 ++++++
 3 files changed, 90 insertions(+), 3 deletions(-)

diff --git a/features/get_experiment_flag.feature b/features/get_experiment_flag.feature
index 45c3eec..fe3bb7d 100644
--- a/features/get_experiment_flag.feature
+++ b/features/get_experiment_flag.feature
@@ -22,6 +22,21 @@ Feature: Get an experiment flag
       | trait | value     |
       | plan  | "premium" |
 
+  Scenario: An identity in a running experiment is exposed to the variant it was served
+    Given the environment has the following features:
+      | feature | enabled | value     |
+      | banner  | true    | "control" |
+    And a running experiment on feature "banner" serving variant "treatment" with value "hello treatment"
+    When the application gets the experiment flag "banner" for identity "user_123" with traits:
+      | trait | value     |
+      | plan  | "premium" |
+    Then the application receives the following flags:
+      | feature | enabled | value             |
+      | banner  | true    | "hello treatment" |
+    And Flagsmith has received the following events:
+      | event          | feature | identifier | value       | traits              |
+      | $flag_exposure | banner  | user_123   | "treatment" | {"plan": "premium"} |
+
   Scenario: An identity outside a running experiment is not exposed
     Given the environment has the following features:
       | feature | enabled | value   |
diff --git a/runner/sdk_spec/flagsmith_admin.py b/runner/sdk_spec/flagsmith_admin.py
index 83ef35a..303ab6e 100644
--- a/runner/sdk_spec/flagsmith_admin.py
+++ b/runner/sdk_spec/flagsmith_admin.py
@@ -7,13 +7,19 @@
 from sdk_spec.http import json_request, wait_until_ok
 
 
+@dataclass
+class Feature:
+    id: int
+    value: Any
+
+
 @dataclass
 class Environment:
     project_id: int
     id: int
     api_key: str
     server_side_key: str
-    features: dict[str, int] = field(default_factory=dict)
+    features: dict[str, Feature] = field(default_factory=dict)
     segments: dict[str, int] = field(default_factory=dict)
 
 
@@ -94,7 +100,51 @@ def create_feature(
             # Flagsmith infers the value type (string, integer or boolean) from `initial_value`.
             body["initial_value"] = value
         feature = self._request("POST", f"/projects/{environment.project_id}/features/", body)
-        environment.features[name] = feature["id"]
+        environment.features[name] = Feature(id=feature["id"], value=value)
+
+    def start_experiment(
+        self, environment: Environment, feature: str, variant: str, value: Any
+    ) -> None:
+        """Runs an experiment on the feature, serving `variant` to every identity.
+
+        A single variant at 100% keeps the scenario deterministic: every identity is
+        enrolled, and every one of them is served that variant.
+        """
+        control = environment.features[feature]
+        option = self._request(
+            "POST",
+            f"/projects/{environment.project_id}/features/{control.id}/mv-options/",
+            {
+                "feature": control.id,
+                "key": variant,
+                # The experiment's rollout sets the allocation this option is served at.
+                "default_percentage_allocation": 0,
+                **_typed_value(value),
+            },
+        )
+        experiment = self._request(
+            "POST",
+            f"/environments/{environment.api_key}/experiments/",
+            {
+                "feature": control.id,
+                "name": f"{feature} experiment",
+                "hypothesis": f"Serving {variant} changes what identities do",
+                "experiment_rollout": {
+                    "enabled": True,
+                    "rollout_percentage": 100,
+                    # What an enrolled identity is served when it is allocated no variant.
+                    "feature_state_value": _rollout_value(control.value),
+                    "multivariate_feature_state_values": [
+                        {"multivariate_feature_option": option["id"], "percentage_allocation": 100}
+                    ],
+                    # No segments: the experiment targets every identity in the environment.
+                    "audience": {"match": "any", "segment_ids": []},
+                },
+            },
+        )
+        self._request(
+            "POST", f"/environments/{environment.api_key}/experiments/{experiment['id']}/start/"
+        )
 
     def create_trait_segment(
         self, environment: Environment, name: str, trait: str, operator: str, value: Any
@@ -130,7 +180,7 @@ def create_segment_override(
         self._request(
             "POST",
             f"/environments/{environment.api_key}/features/"
-            f"{environment.features[feature]}/create-segment-override/",
+            f"{environment.features[feature].id}/create-segment-override/",
             {
                 "enabled": enabled,
                 "feature_state_value": _typed_value(value),
@@ -159,6 +209,15 @@ def _typed_value(value: Any) -> dict[str, Any]:
     return {"type": "unicode", "string_value": None if value is None else str(value)}
 
 
+def _rollout_value(value: Any) -> dict[str, Any]:
+    """An experiment rollout names the value's type differently, and sends it as text."""
+    if isinstance(value, bool):
+        return {"type": "boolean", "value": str(value).lower()}
+    if isinstance(value, int):
+        return {"type": "integer", "value": str(value)}
+    return {"type": "string", "value": "" if value is None else str(value)}
+
+
 def _trait_value(trait: dict[str, Any]) -> Any:
     return {
         "int": trait["integer_value"],
diff --git a/runner/tests/conftest.py b/runner/tests/conftest.py
index 1847d1c..ede0aba 100644
--- a/runner/tests/conftest.py
+++ b/runner/tests/conftest.py
@@ -251,6 +251,19 @@ def environment_features(
         )
 
 
+@given(
+    parsers.parse(
+        'a running experiment on feature "{feature}" serving variant "{variant}" with value {value}'
+    )
+)
+def running_experiment(
+    state: ScenarioState, flagsmith: FlagsmithAdmin, feature: str, variant: str, value: str
+) -> None:
+    flagsmith.start_experiment(
+        require_environment(state), feature, variant, parse_value(value)
+    )
+
+
 @given(parsers.parse('a segment "{segment}" where trait "{trait}" equals {value}'))
 def trait_segment(
     state: ScenarioState, flagsmith: FlagsmithAdmin, segment: str, trait: str, value: str