refactor: pre/post 17 stress workers and SDK integration checks

- Reworks integration coverage by removing the cross-repo subprocess wire-contract test and validating key pop responses directly with `horde_sdk` models inside existing alchemy/controlnet/scheduler tests.
- The annotation migration test now runs the real SQL migration unchanged in an isolated disposable database and uses `sqlparse` for statement splitting, so `sqlparse` is added to dev dependencies.
- The stress suite is restructured to model two image-worker populations (legacy pre-17 and bridge-17+), add separate worker-count/bridge-agent options, and fail fast if legacy workers receive extended sampler payloads. Sampler feature traffic is now generated from SDK sampler constraints with deterministic none/subset/all setting coverage, while related misuse/tagging/docs are updated to match the new workload behavior.

pyproject.toml

@@ -78,6 +78,7 @@ dev = [
     "tox~=4.14.1",
     "pytest-asyncio",
     "pytest-cov",
+    "sqlparse",
     "fakeredis",
     "freezegun",
     "testcontainers[postgres,redis]",

requirements.dev.txt

@@ -9,6 +9,7 @@ tox~=4.14.1
 
 pytest-asyncio
 pytest-cov
+sqlparse
 fakeredis
 freezegun
 testcontainers[postgres,redis]

tests/integration/_sdk_wire_client.py

@@ -1,143 +0,0 @@
-# SPDX-FileCopyrightText: 2026 Tazlin <tazlin@haidra.net>
-#
-# SPDX-License-Identifier: AGPL-3.0-or-later
-
-"""Cross-repo wire-contract client, executed under the local horde_sdk venv.
-
-Reads a JSON job description on stdin, performs real HTTP pops against a live
-AI-Horde server, and validates the responses with the local horde_sdk response
-models. Emits a single-line JSON verdict on stdout. It imports only horde_sdk,
-requests, and the standard library: never the AI-Horde ``horde`` package.
-"""
-
-from __future__ import annotations
-
-import json
-import sys
-
-import requests
-from horde_sdk.ai_horde_api.apimodels.alchemy.pop import AlchemyJobPopResponse, AlchemyPopRequest
-from horde_sdk.ai_horde_api.apimodels.generate.async_ import ImageGenerateAsyncRequest, ImageGenerationInputPayload
-from horde_sdk.ai_horde_api.apimodels.generate.pop import (
-    ImageGenerateJobPopRequest,
-    ImageGenerateJobPopResponse,
-)
-from horde_sdk.generation_parameters.alchemy.consts import KNOWN_ANNOTATION_CONTROL_TYPES
-
-
-def _headers(cfg: dict) -> dict:
-    return {"apikey": cfg["apikey"], "Client-Agent": cfg["client_agent"]}
-
-
-def _check_alchemy(cfg: dict) -> dict:
-    body = cfg["alchemy_pop_body"]
-
-    request_model = AlchemyPopRequest(
-        apikey=cfg["apikey"],
-        name=body["name"],
-        priority_usernames=[],
-        forms=body["forms"],
-        annotation_types=body["annotation_types"],
-        amount=1,
-    )
-    reserialized = request_model.model_dump(by_alias=True, exclude_none=True, mode="json")
-    if reserialized.get("annotation_types") != body["annotation_types"]:
-        raise AssertionError(f"SDK AlchemyPopRequest dropped annotation_types: {reserialized!r}")
-
-    response = requests.post(cfg["base_url"] + "/api/v2/interrogate/pop", json=body, headers=_headers(cfg), timeout=30)
-    response.raise_for_status()
-    raw = response.json()
-
-    parsed = AlchemyJobPopResponse.model_validate(raw)
-    forms = parsed.forms or []
-    if not forms:
-        raise AssertionError(f"alchemy pop returned no forms: {raw!r}")
-    form = forms[0]
-    if form.form != "annotation":
-        raise AssertionError(f"unexpected form: {form.form!r} in {raw!r}")
-    if form.payload is None or form.payload.control_type is None:
-        raise AssertionError(f"SDK did not parse control_type from payload: {raw!r}")
-    if form.payload.control_type != KNOWN_ANNOTATION_CONTROL_TYPES.canny:
-        raise AssertionError(f"control_type did not round-trip: {form.payload.control_type!r}")
-    if not form.r2_upload:
-        raise AssertionError(f"annotation pop missing r2_upload: {raw!r}")
-
-    return {
-        "form": str(form.form),
-        "control_type": str(form.payload.control_type),
-        "control_type_is_enum": isinstance(form.payload.control_type, KNOWN_ANNOTATION_CONTROL_TYPES),
-        "has_r2_upload": bool(form.r2_upload),
-        "request_roundtrip_annotation_types": reserialized.get("annotation_types"),
-    }
-
-
-def _check_image_request(cfg: dict) -> dict:
-    params = ImageGenerationInputPayload(**cfg["image_request_params"])
-    request_model = ImageGenerateAsyncRequest(
-        apikey=cfg["apikey"],
-        prompt="SDK sampler request round trip",
-        params=params,
-        models=["stable_diffusion"],
-    )
-    reserialized = request_model.model_dump(by_alias=True, exclude_none=True, mode="json")["params"]
-    for field, expected in cfg["image_request_params"].items():
-        if reserialized.get(field) != expected:
-            raise AssertionError(f"SDK ImageGenerateAsyncRequest changed {field}: {reserialized!r}")
-    return {field: reserialized[field] for field in cfg["image_request_params"]}
-
-
-def _check_image_pops(cfg: dict) -> dict:
-    popped_payloads = []
-    extended_controlnet_roundtrip = False
-
-    for case in cfg["image_pop_cases"]:
-        body = case["body"]
-        request_model = ImageGenerateJobPopRequest(
-            apikey=cfg["apikey"],
-            name=body["name"],
-            priority_usernames=[],
-            models=body["models"],
-            bridge_agent=body["bridge_agent"],
-            amount=body["amount"],
-            max_pixels=body["max_pixels"],
-            allow_controlnet=body.get("allow_controlnet", False),
-            allow_extended_controlnet=body["allow_extended_controlnet"],
-        )
-        reserialized = request_model.model_dump(by_alias=True, exclude_none=True, mode="json")
-        if reserialized.get("allow_extended_controlnet") is not True:
-            raise AssertionError(f"SDK ImageGenerateJobPopRequest dropped allow_extended_controlnet: {reserialized!r}")
-        extended_controlnet_roundtrip = True
-
-        response = requests.post(cfg["base_url"] + "/api/v2/generate/pop", json=body, headers=_headers(cfg), timeout=30)
-        response.raise_for_status()
-        raw = response.json()
-        parsed = ImageGenerateJobPopResponse.model_validate(raw)
-        if parsed.id_ is None:
-            raise AssertionError(f"image pop matched no job: {raw!r}")
-
-        payload = parsed.payload.model_dump(by_alias=True, exclude_none=True, mode="json")
-        for field, expected in case["expected_payload"].items():
-            if payload.get(field) != expected:
-                raise AssertionError(f"SDK image pop changed {field}: {payload!r}")
-        popped_payloads.append({field: payload[field] for field in case["expected_payload"]})
-
-    return {
-        "payloads": popped_payloads,
-        "request_roundtrip_allow_extended_controlnet": extended_controlnet_roundtrip,
-    }
-
-
-def main() -> int:
-    cfg = json.load(sys.stdin)
-    verdict = {
-        "alchemy": _check_alchemy(cfg),
-        "image_request": _check_image_request(cfg),
-        "image": _check_image_pops(cfg),
-        "ok": True,
-    }
-    sys.stdout.write(json.dumps(verdict))
-    return 0
-
-
-if __name__ == "__main__":
-    sys.exit(main())

tests/integration/test_alchemy.py

@@ -4,6 +4,7 @@
 # SPDX-License-Identifier: AGPL-3.0-or-later
 
 import pytest
+from horde_sdk.ai_horde_api.apimodels import AlchemyJobPopResponse
 
 # The annotation form is image-output: its pop mints an R2 upload URL, so the object store must be
 # provisioned for this module rather than depending on a co-running marked module to bring it up.
@@ -150,6 +151,7 @@ def test_alchemy_annotation(client, request_headers: dict[str, str]) -> None:
 
     assert pop_req.status_code < 400, pop_req.get_data(as_text=True)
     pop_results = pop_req.get_json()
+    sdk_response = AlchemyJobPopResponse.model_validate(pop_results)
 
     popped_form = pop_results["forms"][0]
     job_id = popped_form["id"]
@@ -159,6 +161,9 @@ def test_alchemy_annotation(client, request_headers: dict[str, str]) -> None:
     assert popped_form["r2_upload"], pop_results
     # The parameterized control type must survive round-trip to the worker via the form payload.
     assert popped_form["payload"]["control_type"] == "canny", pop_results
+    assert sdk_response.forms is not None
+    assert sdk_response.forms[0].payload is not None
+    assert sdk_response.forms[0].payload.control_type == "canny"
 
     submit_dict = {
         "id": job_id,

tests/integration/test_annotation_migration.py

@@ -6,42 +6,52 @@
 
 from __future__ import annotations
 
+import re
 import uuid
+from collections.abc import Iterator
 from pathlib import Path
 
 import pytest
 import sqlalchemy
-
-from tests.dependency_runtime import create_schema, drop_schema, new_test_schema_name
+import sqlparse
 
 pytestmark = pytest.mark.integration
 
 
-def _migration_statements(schema_name: str) -> list[str]:
-    repository_root = Path(__file__).parents[2]
-    migration = (repository_root / "sql_statements/5.1.6.txt").read_text(encoding="utf-8")
-    migration = migration.replace("public.workers", f'"{schema_name}".workers').replace(
-        "public.interrogation_worker_annotation_types",
-        f'"{schema_name}".interrogation_worker_annotation_types',
-    )
-    migration = "\n".join(line for line in migration.splitlines() if not line.lstrip().startswith("--"))
-    return [statement.strip() for statement in migration.split(";") if statement.strip()]
+@pytest.fixture
+def migration_database(pg_dsn: str) -> Iterator[str]:
+    """Provide a disposable database so the production migration can run unchanged."""
+    database_name = f"annotation_migration_{uuid.uuid4().hex[:12]}"
+    if re.fullmatch(r"[a-z0-9_]+", database_name) is None:
+        raise RuntimeError(f"Unsafe generated test database name: {database_name!r}")
+
+    admin_engine = sqlalchemy.create_engine(pg_dsn, isolation_level="AUTOCOMMIT")
+    database_url = sqlalchemy.engine.make_url(pg_dsn).set(database=database_name)
+    try:
+        with admin_engine.connect() as connection:
+            connection.exec_driver_sql(f'CREATE DATABASE "{database_name}"')
+        yield database_url.render_as_string(hide_password=False)
+    finally:
+        try:
+            with admin_engine.connect() as connection:
+                connection.exec_driver_sql(f'DROP DATABASE IF EXISTS "{database_name}" WITH (FORCE)')
+        finally:
+            admin_engine.dispose()
 
 
-def _run_migration(connection, schema_name: str) -> None:
-    for statement in _migration_statements(schema_name):
+def _run_migration(connection: sqlalchemy.Connection) -> None:
+    migration_path = Path(__file__).parents[2] / "sql_statements/5.1.6.txt"
+    migration = migration_path.read_text(encoding="utf-8")
+    for statement in sqlparse.split(migration, strip_semicolon=True):
         connection.exec_driver_sql(statement)
 
 
-def test_annotation_capability_migration_upgrades_an_earlier_draft(pg_dsn: str) -> None:
-    schema_name = new_test_schema_name("annotation_migration_test")
-    create_schema(pg_dsn, schema_name)
-    engine = sqlalchemy.create_engine(pg_dsn, isolation_level="AUTOCOMMIT")
+def test_annotation_capability_migration_upgrades_an_earlier_draft(migration_database: str) -> None:
+    engine = sqlalchemy.create_engine(migration_database, isolation_level="AUTOCOMMIT")
     worker_id = uuid.uuid4()
 
     try:
         with engine.connect() as connection:
-            connection.exec_driver_sql(f'SET search_path TO "{schema_name}"')
             connection.exec_driver_sql("CREATE TABLE workers (id UUID PRIMARY KEY)")
             connection.execute(sqlalchemy.text("INSERT INTO workers (id) VALUES (:worker_id)"), {"worker_id": worker_id})
             connection.exec_driver_sql(
@@ -63,8 +73,8 @@ def test_annotation_capability_migration_upgrades_an_earlier_draft(pg_dsn: str)
                 {"worker_id": worker_id},
             )
 
-            _run_migration(connection, schema_name)
-            _run_migration(connection, schema_name)
+            _run_migration(connection)
+            _run_migration(connection)
 
             rows = connection.execute(
                 sqlalchemy.text(
@@ -78,12 +88,11 @@ def test_annotation_capability_migration_upgrades_an_earlier_draft(pg_dsn: str)
                     """
                     SELECT is_nullable
                     FROM information_schema.columns
-                    WHERE table_schema = :schema_name
+                    WHERE table_schema = 'public'
                       AND table_name = 'interrogation_worker_annotation_types'
                       AND column_name = 'annotation_type'
                     """,
                 ),
-                {"schema_name": schema_name},
             ).scalar_one()
             assert column == "NO"
 
@@ -92,10 +101,9 @@ def test_annotation_capability_migration_upgrades_an_earlier_draft(pg_dsn: str)
                     """
                     SELECT indexrelid::regclass::text, indisunique, indisvalid
                     FROM pg_index
-                    WHERE indexrelid = to_regclass(:index_name)
+                    WHERE indexrelid = to_regclass('public.idx_interrogation_worker_annotation_types_worker_type')
                     """,
                 ),
-                {"index_name": f"{schema_name}.idx_interrogation_worker_annotation_types_worker_type"},
             ).one()
             assert index[1:] == (True, True)
 
@@ -114,4 +122,3 @@ def test_annotation_capability_migration_upgrades_an_earlier_draft(pg_dsn: str)
             assert "idx_interrogation_worker_annotation_types_worker_type" in "\n".join(plan)
     finally:
         engine.dispose()
-        drop_schema(pg_dsn, schema_name)

tests/integration/test_image_controlnet_gating.py

@@ -15,6 +15,7 @@ import base64
 from io import BytesIO
 
 import pytest
+from horde_sdk.ai_horde_api.apimodels import ImageGenerateJobPopResponse
 from PIL import Image
 
 from horde.bridge_reference import CAPABILITY_EXPANDED_REGEN_VERSION
@@ -98,9 +99,11 @@ def test_new_control_type_skips_old_bridge_but_matches_new(client, request_heade
         new_pop = client.post("/api/v2/generate/pop", json=_pop_dict(NEW_BRIDGE_AGENT), headers=request_headers)
         assert new_pop.status_code < 400, new_pop.get_data(as_text=True)
         new_results = new_pop.get_json()
+        sdk_response = ImageGenerateJobPopResponse.model_validate(new_results)
         # The extended-capable worker with the opt-in flag matches the same job.
         assert new_results["id"] is not None, new_results
         assert new_results["payload"]["control_type"] == "lineart", new_results
+        assert sdk_response.payload.control_type == "lineart"
     finally:
         client.delete(f"/api/v2/generate/status/{req_id}", headers=request_headers)
 

tests/integration/test_image_scheduler_gating.py

@@ -11,6 +11,7 @@ request reaching the same population of workers it always did.
 """
 
 import pytest
+from horde_sdk.ai_horde_api.apimodels import ImageGenerateJobPopResponse
 
 from horde.bridge_reference import CAPABILITY_EXPANDED_REGEN_VERSION
 
@@ -85,8 +86,10 @@ def test_extended_schedule_skips_old_bridge_but_matches_new(
         new_pop = client.post("/api/v2/generate/pop", json=_pop_dict(NEW_BRIDGE_AGENT), headers=request_headers)
         assert new_pop.status_code < 400, new_pop.get_data(as_text=True)
         new_results = new_pop.get_json()
+        sdk_response = ImageGenerateJobPopResponse.model_validate(new_results)
         assert new_results["id"] is not None, new_results
         assert new_results["payload"]["scheduler"] == schedule, new_results
+        assert sdk_response.payload.scheduler == schedule
     finally:
         client.delete(f"/api/v2/generate/status/{req_id}", headers=request_headers)
 
@@ -202,8 +205,10 @@ def test_solver_option_skips_old_bridge_but_matches_new(
         new_pop = client.post("/api/v2/generate/pop", json=_pop_dict(NEW_BRIDGE_AGENT), headers=request_headers)
         assert new_pop.status_code < 400, new_pop.get_data(as_text=True)
         new_results = new_pop.get_json()
+        sdk_response = ImageGenerateJobPopResponse.model_validate(new_results)
         assert new_results["id"] is not None, new_results
         assert new_results["payload"][field] == value, new_results
+        assert getattr(sdk_response.payload, field) == value
     finally:
         client.delete(f"/api/v2/generate/status/{req_id}", headers=request_headers)
 
@@ -230,8 +235,10 @@ def test_flow_shift_skips_old_bridge_but_matches_new(client, request_headers: di
         new_pop = client.post("/api/v2/generate/pop", json=new_pop_body, headers=request_headers)
         assert new_pop.status_code < 400, new_pop.get_data(as_text=True)
         new_results = new_pop.get_json()
+        sdk_response = ImageGenerateJobPopResponse.model_validate(new_results)
         assert new_results["id"] is not None, new_results
         assert new_results["payload"]["flow_shift"] == 1.1, new_results
+        assert sdk_response.payload.flow_shift == 1.1
     finally:
         client.delete(f"/api/v2/generate/status/{req_id}", headers=request_headers)
 

tests/integration/test_sdk_wire_contract.py

@@ -1,258 +0,0 @@
-# SPDX-FileCopyrightText: 2026 Tazlin <tazlin@haidra.net>
-#
-# SPDX-License-Identifier: AGPL-3.0-or-later
-
-"""Cross-repo wire-contract proof: live HTTP server parsed by the local horde_sdk.
-
-The unit/integration suites exercise the server in isolation. This module proves the wire contract
-between the AI-Horde server tree and the horde_sdk tree: real HTTP pop responses parse with the new
-annotation, ControlNet, sampler, scheduler, solver and flow-shift fields intact, and SDK request models
-re-serialize the same fields. The SDK runs as a subprocess so the server and client never share an
-interpreter.
-"""
-
-from __future__ import annotations
-
-import base64
-import json
-import os
-import socket
-import subprocess
-import threading
-from collections.abc import Iterator
-from io import BytesIO
-from pathlib import Path
-
-import pytest
-from PIL import Image
-from werkzeug.serving import make_server
-
-from horde.bridge_reference import CAPABILITY_EXPANDED_REGEN_VERSION
-
-# The SDK-side interpreter is host-specific: point HORDE_SDK_WIRE_PYTHON at a python whose
-# environment has the horde_sdk under test installed; the suite skips when it is not set.
-SDK_PYTHON = Path(os.environ["HORDE_SDK_WIRE_PYTHON"]) if os.environ.get("HORDE_SDK_WIRE_PYTHON") else None
-SDK_CLIENT_SCRIPT = Path(__file__).resolve().parent / "_sdk_wire_client.py"
-
-TEST_MODELS = ["stable_diffusion"]
-SDXL_MODELS = ["AlbedoBase XL (SDXL)"]
-FLOW_MODELS = ["Flux.1-Schnell fp8 (Compact)"]
-NEW_BRIDGE_AGENT = f"AI Horde Worker reGen:{CAPABILITY_EXPANDED_REGEN_VERSION}:https://github.com/Haidra-Org/horde-worker-reGen"
-
-pytestmark = [
-    pytest.mark.object_storage,
-    pytest.mark.usefixtures("object_store_ready"),
-    pytest.mark.skipif(
-        SDK_PYTHON is None or not SDK_PYTHON.exists(),
-        reason="HORDE_SDK_WIRE_PYTHON not set or does not exist",
-    ),
-]
-
-
-def _load_image_as_b64(image_path: str) -> str:
-    final_src_img = Image.open(image_path)
-    buffer = BytesIO()
-    final_src_img.save(buffer, format="Webp", quality=50, exact=True)
-    return base64.b64encode(buffer.getvalue()).decode("utf8")
-
-
-@pytest.fixture
-def live_server(app) -> Iterator[str]:
-    probe = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-    probe.bind(("127.0.0.1", 0))
-    port = probe.getsockname()[1]
-    probe.close()
-
-    server = make_server("127.0.0.1", port, app, threaded=True)
-    thread = threading.Thread(target=server.serve_forever, daemon=True)
-    thread.start()
-    try:
-        yield f"http://127.0.0.1:{port}"
-    finally:
-        server.shutdown()
-        thread.join(timeout=5)
-
-
-def test_sdk_parses_live_pop_responses(
-    client,
-    live_server: str,
-    api_key: str,
-    request_headers: dict[str, str],
-    settle_kudos,
-) -> None:
-    settle_kudos()
-    annotation_async = {
-        "forms": [{"name": "annotation", "payload": {"control_type": "canny"}}],
-        "source_image": "https://github.com/Haidra-Org/AI-Horde/blob/main/icon.png?raw=true",
-    }
-    annotation_req = client.post("/api/v2/interrogate/async", json=annotation_async, headers=request_headers)
-    assert annotation_req.status_code < 400, annotation_req.get_data(as_text=True)
-    annotation_id = annotation_req.get_json()["id"]
-
-    image_async = {
-        "prompt": "a controlnet-guided horde of robots",
-        "nsfw": True,
-        "censor_nsfw": False,
-        "r2": True,
-        "shared": True,
-        "trusted_workers": True,
-        "params": {
-            "width": 512,
-            "height": 512,
-            "steps": 20,
-            "cfg_scale": 7.5,
-            "sampler_name": "k_euler",
-            "scheduler": "exponential",
-            "sampler_s_noise": 1.0,
-            "sampler_s_churn": 0.1,
-            "sampler_s_tmin": 0.0,
-            "sampler_s_tmax": 1.0,
-            "control_type": "lineart",
-        },
-        "models": TEST_MODELS,
-        "source_image": _load_image_as_b64("img_stable/0.jpg"),
-        "source_processing": "img2img",
-    }
-    image_req = client.post("/api/v2/generate/async", json=image_async, headers=request_headers)
-    assert image_req.status_code < 400, image_req.get_data(as_text=True)
-    image_ids = [image_req.get_json()["id"]]
-
-    order_req = client.post(
-        "/api/v2/generate/async",
-        json={
-            "prompt": "an SDK order-field wire probe",
-            "params": {
-                "width": 512,
-                "height": 512,
-                "steps": 20,
-                "sampler_name": "k_lms",
-                "scheduler": "normal",
-                "sampler_order": 4,
-            },
-            "models": SDXL_MODELS,
-        },
-        headers=request_headers,
-    )
-    assert order_req.status_code < 400, order_req.get_data(as_text=True)
-    image_ids.append(order_req.get_json()["id"])
-
-    flow_req = client.post(
-        "/api/v2/generate/async",
-        json={
-            "prompt": "an SDK flow-field wire probe",
-            "params": {
-                "width": 512,
-                "height": 512,
-                "steps": 20,
-                "cfg_scale": 1,
-                "sampler_name": "exp_heun_2_x0_sde",
-                "scheduler": "normal",
-                "sampler_eta": 1.0,
-                "sampler_s_noise": 1.0,
-                "sampler_solver_type": "phi_1",
-                "flow_shift": 1.1,
-            },
-            "models": FLOW_MODELS,
-        },
-        headers=request_headers,
-    )
-    assert flow_req.status_code < 400, flow_req.get_data(as_text=True)
-    image_ids.append(flow_req.get_json()["id"])
-
-    cfg = {
-        "base_url": live_server,
-        "apikey": api_key,
-        "client_agent": request_headers["Client-Agent"],
-        "alchemy_pop_body": {
-            "name": "CICD Fake Alchemist",
-            "forms": ["annotation"],
-            "annotation_types": ["canny"],
-            "bridge_agent": request_headers["Client-Agent"],
-            "max_tiles": 96,
-        },
-        "image_request_params": {
-            "sampler_name": "exp_heun_2_x0_sde",
-            "scheduler": "normal",
-            "sampler_eta": 1.0,
-            "sampler_s_noise": 1.0,
-            "sampler_s_churn": 0.1,
-            "sampler_s_tmin": 0.0,
-            "sampler_s_tmax": 1.0,
-            "sampler_solver_type": "phi_1",
-            "sampler_order": 4,
-            "flow_shift": 1.1,
-        },
-        "image_pop_cases": [],
-    }
-    for models, expected_payload in [
-        (
-            TEST_MODELS,
-            {
-                "sampler_name": "k_euler",
-                "scheduler": "exponential",
-                "sampler_s_noise": 1.0,
-                "sampler_s_churn": 0.1,
-                "sampler_s_tmin": 0.0,
-                "sampler_s_tmax": 1.0,
-                "control_type": "lineart",
-            },
-        ),
-        (SDXL_MODELS, {"sampler_name": "k_lms", "scheduler": "normal", "sampler_order": 4}),
-        (
-            FLOW_MODELS,
-            {
-                "sampler_name": "exp_heun_2_x0_sde",
-                "scheduler": "normal",
-                "sampler_eta": 1.0,
-                "sampler_s_noise": 1.0,
-                "sampler_solver_type": "phi_1",
-                "flow_shift": 1.1,
-            },
-        ),
-    ]:
-        cfg["image_pop_cases"].append(
-            {
-                "body": {
-                    "name": "CICD Fake Dreamer",
-                    "models": models,
-                    "bridge_agent": NEW_BRIDGE_AGENT,
-                    "nsfw": True,
-                    "amount": 10,
-                    "max_pixels": 4194304,
-                    "allow_img2img": True,
-                    "allow_painting": True,
-                    "allow_unsafe_ipaddr": True,
-                    "allow_post_processing": True,
-                    "allow_controlnet": True,
-                    "allow_extended_controlnet": True,
-                    "allow_sdxl_controlnet": True,
-                    "allow_lora": True,
-                },
-                "expected_payload": expected_payload,
-            },
-        )
-
-    try:
-        completed = subprocess.run(
-            [str(SDK_PYTHON), str(SDK_CLIENT_SCRIPT)],
-            input=json.dumps(cfg),
-            capture_output=True,
-            text=True,
-            cwd=str(SDK_CLIENT_SCRIPT.parent),
-            timeout=120,
-        )
-        assert completed.returncode == 0, (
-            f"SDK wire client failed (rc={completed.returncode}).\nSTDOUT:\n{completed.stdout}\nSTDERR:\n{completed.stderr}"
-        )
-        verdict = json.loads(completed.stdout.strip().splitlines()[-1])
-        assert verdict["ok"] is True, verdict
-        assert verdict["alchemy"]["control_type"] == "canny", verdict
-        assert verdict["alchemy"]["has_r2_upload"] is True, verdict
-        assert verdict["alchemy"]["request_roundtrip_annotation_types"] == ["canny"], verdict
-        assert verdict["image"]["payloads"] == [case["expected_payload"] for case in cfg["image_pop_cases"]], verdict
-        assert verdict["image"]["request_roundtrip_allow_extended_controlnet"] is True, verdict
-        assert verdict["image_request"] == cfg["image_request_params"], verdict
-    finally:
-        client.delete(f"/api/v2/interrogate/status/{annotation_id}", headers=request_headers)
-        for image_id in image_ids:
-            client.delete(f"/api/v2/generate/status/{image_id}", headers=request_headers)

tests/stress/README.md

@@ -111,16 +111,21 @@ cd tests/stress && locust
 locust -f tests/stress/locustfile.py --host http://localhost:7001
 ```
 
-The mixed workload includes a `SamplerFeatureRequester`. With
-`--sampler-feature-requestors 1`, it deterministically rotates through every
-new sampler plus all explicit schedulers, solver-control fields, and
-`flow_shift`. It keeps at most eight requests queued and cancels older ones, so
-the sweep exercises request validation, waiting-prompt creation, worker
-matching, and cancellation without creating an unbounded backlog. Meta users
-also fetch `/api/v2/status/sampler_constraints`, while misuse users exercise
-the new hard-constraint rejection paths. The default model pool includes the
-SD1 and Flux models needed by the baseline-specific cases, and simulated image
-workers advertise reGen bridge 17 capabilities by default.
+The mixed workload contains separate pre-17 and bridge-17+ image worker
+populations. Pre-17 workers actively fail the Locust run if they receive an
+extended sampler, an explicit scheduler, a solver-control field, or
+`flow_shift`; bridge-17+ workers serve those jobs. Both populations always
+offer `stable_diffusion`, and the newer population also offers the default Flux
+model, while their remaining model sets vary normally.
+
+`SamplerFeatureRequester` walks every extended sampler using no optional
+settings and every applicable setting. Where at least two settings apply, it
+also selects a non-empty proper subset; the same zero/subset/all pattern covers
+the Flux `flow_shift` profile. Applicable solver controls come from the installed
+`horde_sdk` constraints rather than a second Locust-owned compatibility table.
+Use `--sampler-feature-requestors 1 --legacy-image-workers 1
+--extended-image-workers 1` for a deterministic smoke population. Meta users
+also fetch `/api/v2/status/sampler_constraints` as ordinary read-only traffic.
 
 Staged load profile:
 
@@ -401,16 +406,14 @@ A fourth job, `stress-kudos-ledger-job`, does not fit this checker pattern: it
 starts a second `--quorum` server and gates inside the harnesses themselves. See
 the CI regression job note under Kudos ledger operational verification above.
 
-The attribution job runs its server with `HORDE_TEST_RATELIMIT_DISABLED=1`: the
+The attribution and staged sampler-sweep jobs run their server with
+`HORDE_TEST_RATELIMIT_DISABLED=1`: the
 oracle probes pop/declare and maintenance interleavings rather than rate-limit
 behaviour, and a throttled run could reach zero violations without ever reaching
 the interleavings the gate is meant to protect. It passes `--stats` to the
 checker for the same reason, so a run that drove no requests fails instead of
 reporting a vacuous pass. The classic smoke job leaves the limiter in force,
-since the suite counts 429s as successes and the gate is crash-class only. The
-staged-shape smoke disables it so the pinned sampler-feature user can complete
-its deterministic catalog; rate-limit behavior remains covered by the classic
-job.
+since the suite counts 429s as successes and the gate is crash-class only.
 
 The hot-user convoy and queue-pressure scenarios are not in CI. Their verdicts
 come from `analyze_hot_user_convoy.py` and `analyze_queue_pressure.py`, which are

tests/stress/locust.conf.example

@@ -71,9 +71,8 @@ spawn-rate = 5
 #
 # Models:
 #   --horde-models / HORDE_MODELS
-#       Comma-separated model names (default also includes stable_diffusion and
-#       Flux.1-Schnell fp8 (Compact), which cover baseline-specific scheduler
-#       and flow-shift cases).
+#       Comma-separated model names (default includes stable_diffusion and
+#       Flux.1-Schnell fp8 (Compact) for the flow-shift profile).
 #
 # Identity:
 #   --client-agent / HORDE_CLIENT_AGENT
@@ -92,14 +91,19 @@ spawn-rate = 5
 #
 # Worker simulation:
 #   --worker-max-pixels / HORDE_WORKER_MAX_PIXELS         (default: 4194304)
-#   --worker-bridge-agent / HORDE_WORKER_BRIDGE_AGENT     (default: AI Horde Worker reGen:17.0.0:...)
+#   --worker-bridge-agent / HORDE_WORKER_BRIDGE_AGENT     (default: AI Horde Worker reGen:17.0.0-stress:...)
+#   --legacy-worker-bridge-agent / HORDE_LEGACY_WORKER_BRIDGE_AGENT
+#                                                         (default: AI Horde Worker reGen:16.0.0-stress:...)
+#   --legacy-image-workers / HORDE_LEGACY_IMAGE_WORKERS   (default: 0 / weight-based)
+#   --extended-image-workers / HORDE_EXTENDED_IMAGE_WORKERS
+#                                                         (default: 0 / weight-based)
 #   --sim-gen-time-min / HORDE_SIM_GEN_TIME_MIN           (default: 4.0 seconds)
 #   --sim-gen-time-max / HORDE_SIM_GEN_TIME_MAX           (default: 120.0 seconds)
 #
 # Targeted feature coverage:
 #   --sampler-feature-requestors / HORDE_SAMPLER_FEATURE_REQUESTORS (default: 0)
-#       Fixed SamplerFeatureRequester users. CI sets this to 1 so every new
-#       sampler, scheduler, solver setting, and flow_shift is exercised.
+#       Cycles every extended sampler through zero/all applicable settings and
+#       proper subsets where possible; also covers Flux flow_shift.
 #
 # User behavior:
 #   --anon-chance-poller / HORDE_ANON_CHANCE_POLLER       (default: 0.75)

tests/stress/locustfile.py

@@ -17,9 +17,11 @@ For staged load profiles, use ``locustfile_shaped.py`` instead.
 """
 
 from locustsuite import (
+    ExtendedWorkerSimulator,
     HotPathRequester,
     InterrogationRequester,
     InterrogationWorkerSimulator,
+    LegacyWorkerSimulator,
     MetaBrowser,
     MisuseUser,
     RequestGenerator,
@@ -27,13 +29,14 @@ from locustsuite import (
     StatusPoller,
     TextRequester,
     TextWorkerSimulator,
-    WorkerSimulator,
 )
 
 __all__ = [
+    "ExtendedWorkerSimulator",
     "HotPathRequester",
     "InterrogationRequester",
     "InterrogationWorkerSimulator",
+    "LegacyWorkerSimulator",
     "MetaBrowser",
     "MisuseUser",
     "RequestGenerator",
@@ -41,5 +44,4 @@ __all__ = [
     "StatusPoller",
     "TextRequester",
     "TextWorkerSimulator",
-    "WorkerSimulator",
 ]

tests/stress/locustfile_shaped.py

@@ -10,9 +10,11 @@ This imports the same users/events as ``locustfile.py`` and additionally exposes
 """
 
 from locustsuite import (
+    ExtendedWorkerSimulator,
     HotPathRequester,
     InterrogationRequester,
     InterrogationWorkerSimulator,
+    LegacyWorkerSimulator,
     MetaBrowser,
     MisuseUser,
     RequestGenerator,
@@ -20,15 +22,16 @@ from locustsuite import (
     StatusPoller,
     TextRequester,
     TextWorkerSimulator,
-    WorkerSimulator,
 )
 from locustsuite.shapes import HordeStagesShape
 
 __all__ = [
+    "ExtendedWorkerSimulator",
     "HordeStagesShape",
     "HotPathRequester",
     "InterrogationRequester",
     "InterrogationWorkerSimulator",
+    "LegacyWorkerSimulator",
     "MetaBrowser",
     "MisuseUser",
     "RequestGenerator",
@@ -36,5 +39,4 @@ __all__ = [
     "StatusPoller",
     "TextRequester",
     "TextWorkerSimulator",
-    "WorkerSimulator",
 ]

tests/stress/locustsuite/__init__.py

@@ -10,9 +10,11 @@ the concrete ``HttpUser`` classes that Locust should discover.
 
 from . import events as _events  # noqa: F401 - imported for Locust event registration side effects
 from .users import (
+    ExtendedWorkerSimulator,
     HotPathRequester,
     InterrogationRequester,
     InterrogationWorkerSimulator,
+    LegacyWorkerSimulator,
     MetaBrowser,
     MisuseUser,
     RequestGenerator,
@@ -20,13 +22,14 @@ from .users import (
     StatusPoller,
     TextRequester,
     TextWorkerSimulator,
-    WorkerSimulator,
 )
 
 __all__ = [
+    "ExtendedWorkerSimulator",
     "HotPathRequester",
     "InterrogationRequester",
     "InterrogationWorkerSimulator",
+    "LegacyWorkerSimulator",
     "MetaBrowser",
     "MisuseUser",
     "RequestGenerator",
@@ -34,5 +37,4 @@ __all__ = [
     "StatusPoller",
     "TextRequester",
     "TextWorkerSimulator",
-    "WorkerSimulator",
 ]

tests/stress/locustsuite/config.py

@@ -37,10 +37,11 @@ _HOT_PROMPT = "a serene cyberpunk landscape at sunset, ultra detailed"
 _HOT_TEXT_PROMPT = "Once upon a time in a faraway land,"
 _INTERROGATION_FORMS = ["caption", "interrogation", "nsfw", "vectorize", "palette", "describe", "aesthetic"]
 
-# Keep this list local to the black-box workload instead of importing the Horde
-# application. Importing the app from a locustfile initializes server-side
-# configuration and defeats the point of exercising a separately running target.
-_NEW_IMAGE_SAMPLERS = (
+# The bridge-17 sampler tiers accepted by this API. The SDK supplies the
+# per-sampler setting constraints; this short list defines which API additions
+# the stress workload should exercise and which jobs a pre-17 worker must never
+# receive.
+_EXTENDED_IMAGE_SAMPLERS = (
     "uni_pc",
     "uni_pc_bh2",
     "dpmpp_2m_sde",
@@ -70,66 +71,16 @@ _NEW_IMAGE_SAMPLERS = (
     "sa_solver_pece",
 )
 
-# Deterministic cases for SamplerFeatureRequester. The first block makes every
-# newly accepted sampler traverse async creation and worker matching. The
-# remaining cases cover every explicit scheduler and solver-control field with
-# combinations accepted by the public constraints document.
-_IMAGE_SAMPLER_FEATURE_CASES = tuple(
+_EXTENDED_SAMPLER_SETTING_FIELDS = frozenset(
     {
-        "name": f"sampler-{sampler}",
-        "params": {"sampler_name": sampler, "scheduler": "karras"},
-    }
-    for sampler in _NEW_IMAGE_SAMPLERS
-) + (
-    {"name": "scheduler-normal", "params": {"sampler_name": "uni_pc", "scheduler": "normal"}},
-    {"name": "scheduler-simple", "params": {"sampler_name": "uni_pc", "scheduler": "simple"}},
-    {"name": "scheduler-sgm-uniform", "params": {"sampler_name": "uni_pc", "scheduler": "sgm_uniform"}},
-    {"name": "scheduler-exponential", "params": {"sampler_name": "uni_pc", "scheduler": "exponential"}},
-    {"name": "scheduler-ddim-uniform", "params": {"sampler_name": "uni_pc", "scheduler": "ddim_uniform"}},
-    {"name": "scheduler-beta", "params": {"sampler_name": "uni_pc", "scheduler": "beta"}},
-    {
-        "name": "scheduler-linear-quadratic",
-        "params": {"sampler_name": "uni_pc", "scheduler": "linear_quadratic"},
-    },
-    {"name": "scheduler-kl-optimal", "params": {"sampler_name": "uni_pc", "scheduler": "kl_optimal"}},
-    {
-        "name": "scheduler-align-your-steps",
-        "params": {"sampler_name": "k_euler", "scheduler": "align_your_steps"},
-        "models": ("stable_diffusion",),
-    },
-    {
-        "name": "scheduler-gits",
-        "params": {"sampler_name": "k_euler", "scheduler": "gits"},
-        "models": ("stable_diffusion",),
-    },
-    {
-        "name": "solver-eta-noise-type",
-        "params": {
-            "sampler_name": "dpmpp_2m_sde",
-            "scheduler": "beta",
-            "sampler_eta": 0.8,
-            "sampler_s_noise": 1.05,
-            "sampler_solver_type": "heun",
-        },
-    },
-    {
-        "name": "solver-churn-window",
-        "params": {
-            "sampler_name": "k_euler",
-            "scheduler": "exponential",
-            "sampler_s_churn": 0.1,
-            "sampler_s_noise": 1.05,
-            "sampler_s_tmin": 0.0,
-            "sampler_s_tmax": 10.0,
-        },
-    },
-    {
-        "name": "solver-order",
-        "params": {"sampler_name": "deis", "scheduler": "sgm_uniform", "sampler_order": 3},
-    },
-    {
-        "name": "flow-shift",
-        "params": {"sampler_name": "k_euler", "scheduler": "simple", "flow_shift": 1.1},
-        "models": ("Flux.1-Schnell fp8 (Compact)",),
+        "scheduler",
+        "sampler_eta",
+        "sampler_s_noise",
+        "sampler_s_churn",
+        "sampler_s_tmin",
+        "sampler_s_tmax",
+        "sampler_solver_type",
+        "sampler_order",
+        "flow_shift",
     },
 )

tests/stress/locustsuite/events.py

@@ -125,8 +125,15 @@ def add_ai_horde_arguments(parser):
         "--worker-bridge-agent",
         type=str,
         env_var="HORDE_WORKER_BRIDGE_AGENT",
-        default="AI Horde Worker reGen:17.0.0:https://github.com/Haidra-Org/horde-worker-reGen",
-        help="Bridge agent string for simulated workers",
+        default="AI Horde Worker reGen:17.0.0-stress:https://github.com/Haidra-Org/horde-worker-reGen",
+        help="Bridge agent string for bridge-17+ simulated workers",
+    )
+    group.add_argument(
+        "--legacy-worker-bridge-agent",
+        type=str,
+        env_var="HORDE_LEGACY_WORKER_BRIDGE_AGENT",
+        default="AI Horde Worker reGen:16.0.0-stress:https://github.com/Haidra-Org/horde-worker-reGen",
+        help="Bridge agent string for pre-17 simulated workers",
     )
     group.add_argument(
         "--sim-gen-time-min",
@@ -204,6 +211,8 @@ def add_ai_horde_arguments(parser):
     for _flag, _envvar, _help in (
         ("--image-requestors", "HORDE_IMAGE_REQUESTORS", "Concurrent RequestGenerator users (image POST /generate/async)"),
         ("--image-workers", "HORDE_IMAGE_WORKERS", "Concurrent WorkerSimulator users (image POST /generate/pop+submit)"),
+        ("--legacy-image-workers", "HORDE_LEGACY_IMAGE_WORKERS", "Concurrent pre-17 image worker users"),
+        ("--extended-image-workers", "HORDE_EXTENDED_IMAGE_WORKERS", "Concurrent bridge-17+ image worker users"),
         ("--text-requestors", "HORDE_TEXT_REQUESTORS", "Concurrent TextRequester users"),
         ("--text-workers", "HORDE_TEXT_WORKERS", "Concurrent TextWorkerSimulator users"),
         ("--interrogate-requestors", "HORDE_INTERROGATE_REQUESTORS", "Concurrent InterrogationRequester users"),
@@ -217,7 +226,7 @@ def add_ai_horde_arguments(parser):
         (
             "--sampler-feature-requestors",
             "HORDE_SAMPLER_FEATURE_REQUESTORS",
-            "Concurrent SamplerFeatureRequester users (new samplers, schedulers, and solver settings)",
+            "Concurrent sampler feature requestors (zero, subset, and all extended settings)",
         ),
         ("--meta-browsers", "HORDE_META_BROWSERS", "Concurrent MetaBrowser users (read-only meta endpoints)"),
         ("--misuse-users", "HORDE_MISUSE_USERS", "Concurrent MisuseUser users (4xx validation paths)"),
@@ -428,9 +437,11 @@ def on_test_start(environment, **kw):
     # spawner respects it. We do this here rather than at import time so the
     # CLI arguments are guaranteed to be parsed.
     from .users import (
+        ExtendedWorkerSimulator,
         HotPathRequester,
         InterrogationRequester,
         InterrogationWorkerSimulator,
+        LegacyWorkerSimulator,
         MetaBrowser,
         MisuseUser,
         RequestGenerator,
@@ -438,13 +449,13 @@ def on_test_start(environment, **kw):
         StatusPoller,
         TextRequester,
         TextWorkerSimulator,
-        WorkerSimulator,
     )
 
     _fixed_count_overrides = (
         (StatusPoller, "status_pollers"),
         (RequestGenerator, "image_requestors"),
-        (WorkerSimulator, "image_workers"),
+        (LegacyWorkerSimulator, "legacy_image_workers"),
+        (ExtendedWorkerSimulator, "extended_image_workers"),
         (TextRequester, "text_requestors"),
         (TextWorkerSimulator, "text_workers"),
         (InterrogationRequester, "interrogate_requestors"),
@@ -454,6 +465,15 @@ def on_test_start(environment, **kw):
         (MetaBrowser, "meta_browsers"),
         (MisuseUser, "misuse_users"),
     )
+    total_image_workers = int(getattr(opts, "image_workers", 0) or 0)
+    if total_image_workers and not opts.legacy_image_workers and not opts.extended_image_workers:
+        if total_image_workers < 2:
+            logger.warning("--image-workers=1 cannot represent both bridge generations; using one bridge-17+ worker")
+            opts.extended_image_workers = 1
+        else:
+            opts.legacy_image_workers = total_image_workers // 2
+            opts.extended_image_workers = total_image_workers - opts.legacy_image_workers
+
     fixed_total = 0
     for cls, attr in _fixed_count_overrides:
         n = int(getattr(opts, attr, 0) or 0)

tests/stress/locustsuite/users/__init__.py

@@ -4,16 +4,25 @@
 
 """User classes exported for Locust discovery."""
 
-from .image import HotPathRequester, RequestGenerator, SamplerFeatureRequester, StatusPoller, WorkerSimulator
+from .image import (
+    ExtendedWorkerSimulator,
+    HotPathRequester,
+    LegacyWorkerSimulator,
+    RequestGenerator,
+    SamplerFeatureRequester,
+    StatusPoller,
+)
 from .interrogation import InterrogationRequester, InterrogationWorkerSimulator
 from .meta import MetaBrowser
 from .misuse import MisuseUser
 from .text import TextRequester, TextWorkerSimulator
 
 __all__ = [
+    "ExtendedWorkerSimulator",
     "HotPathRequester",
     "InterrogationRequester",
     "InterrogationWorkerSimulator",
+    "LegacyWorkerSimulator",
     "MetaBrowser",
     "MisuseUser",
     "RequestGenerator",
@@ -21,5 +30,4 @@ __all__ = [
     "StatusPoller",
     "TextRequester",
     "TextWorkerSimulator",
-    "WorkerSimulator",
 ]

tests/stress/locustsuite/users/image.py

@@ -10,10 +10,17 @@ import time
 from collections import deque
 from itertools import cycle
 
+from horde_sdk.generation_parameters.image.constraints import SAMPLER_CONSTRAINTS
 from locust import HttpUser, between, tag, task
 from locust.exception import RescheduleTask
 
-from ..config import _EXPECTED_RC_RECOVER, _HOT_PROMPT, _IMAGE_SAMPLER_FEATURE_CASES, _config
+from ..config import (
+    _EXPECTED_RC_RECOVER,
+    _EXTENDED_IMAGE_SAMPLERS,
+    _EXTENDED_SAMPLER_SETTING_FIELDS,
+    _HOT_PROMPT,
+    _config,
+)
 from ..helpers import (
     _handle_async_generate,
     _headers,
@@ -26,8 +33,6 @@ from ..helpers import (
     _safe_json,
 )
 
-_image_sampler_feature_cases = cycle(_IMAGE_SAMPLER_FEATURE_CASES)
-
 
 class StatusPoller(HttpUser):
     """Simulates clients polling /generate/check and /generate/status.
@@ -370,12 +375,14 @@ class WorkerSimulator(HttpUser):
     start_generation, set_generation (record_contribution, R2, webhook).
     """
 
-    weight = 2
-    fixed_count = 0  # set via --image-workers in on_test_start
+    abstract = True
+    bridge_generation = "base"
+    supports_extended_samplers = False
+    required_models: tuple[str, ...] = ()
     wait_time = between(1, 4)
 
     def create_worker_name(self):
-        return f"StressWorker-{''.join(random.choices(string.ascii_lowercase, k=4))}"
+        return f"StressWorker-{self.bridge_generation}-{''.join(random.choices(string.ascii_lowercase, k=4))}"
 
     def on_start(self):
         self.worker_name = self.create_worker_name()
@@ -383,6 +390,7 @@ class WorkerSimulator(HttpUser):
 
         models = _config.get("models", [])
         self.worker_models = random.sample(models, k=random.randint(1, max(1, len(models))))
+        self.worker_models.extend(model for model in self.required_models if model in models and model not in self.worker_models)
 
         # Check that this worker doesn't already exist (from a previous test run), and choose a new name if so.
         # The endpoint returns 200 + worker JSON if a worker by that name exists,
@@ -446,7 +454,7 @@ class WorkerSimulator(HttpUser):
         pop_payload = {
             "name": self.worker_name,
             "models": self.worker_models,
-            "bridge_agent": opts.worker_bridge_agent,
+            "bridge_agent": self.bridge_agent,
             "nsfw": True,
             "amount": 1,
             "max_pixels": opts.worker_max_pixels,
@@ -455,6 +463,7 @@ class WorkerSimulator(HttpUser):
             "allow_unsafe_ipaddr": True,
             "allow_post_processing": True,
             "allow_controlnet": True,
+            "allow_extended_controlnet": self.supports_extended_samplers,
             "allow_lora": True,
         }
         with self.client.post(
@@ -496,6 +505,12 @@ class WorkerSimulator(HttpUser):
             if not job_id:
                 resp.success()
                 return
+            incompatibilities = self._extended_payload_features(pop_data.get("payload") or {})
+            if not self.supports_extended_samplers and incompatibilities:
+                resp.failure(
+                    "Pre-17 worker received an incompatible job containing: " + ", ".join(incompatibilities),
+                )
+                return
             resp.success()
 
         time.sleep(random.uniform(opts.sim_gen_time_min, opts.sim_gen_time_max))
@@ -534,24 +549,107 @@ class WorkerSimulator(HttpUser):
                 return
             resp.failure(f"Submit failed: {resp.status_code}: {resp.text[:200]}")
 
+    @staticmethod
+    def _extended_payload_features(payload: dict) -> list[str]:
+        """Return bridge-17-only features present in a popped job payload."""
+        features = sorted(_EXTENDED_SAMPLER_SETTING_FIELDS.intersection(payload))
+        sampler_name = payload.get("sampler_name")
+        if sampler_name in _EXTENDED_IMAGE_SAMPLERS:
+            features.insert(0, f"sampler_name={sampler_name}")
+        return features
 
-class SamplerFeatureRequester(HttpUser):
-    """Continuously cover the current sampler, scheduler, and solver-control API.
 
-    CI pins one instance of this user so a smoke run walks the cases in a
-    deterministic cycle. Only a small number of requests remain queued at once;
-    older ones are cancelled so the coverage sweep does not overwhelm the fake
-    workers or consume an unbounded amount of requestor kudos.
-    """
+class LegacyWorkerSimulator(WorkerSimulator):
+    """A pre-17 worker, used to verify dispatch excludes extended jobs."""
 
     weight = 1
-    fixed_count = 0  # set via --sampler-feature-requestors in on_test_start
+    fixed_count = 0
+    bridge_generation = "pre17"
+    required_models = ("stable_diffusion",)
+
+    def on_start(self):
+        self.bridge_agent = self.environment.parsed_options.legacy_worker_bridge_agent
+        super().on_start()
+
+
+class ExtendedWorkerSimulator(WorkerSimulator):
+    """A bridge-17+ worker capable of consuming extended sampler jobs."""
+
+    weight = 1
+    fixed_count = 0
+    bridge_generation = "17plus"
+    supports_extended_samplers = True
+    required_models = ("stable_diffusion", "Flux.1-Schnell fp8 (Compact)")
+
+    def on_start(self):
+        self.bridge_agent = self.environment.parsed_options.worker_bridge_agent
+        super().on_start()
+
+
+_SAMPLER_SETTING_VALUES = {
+    "eta": 0.8,
+    "s_noise": 1.05,
+    "s_churn": 0.1,
+    "s_tmin": 0.0,
+    "s_tmax": 10.0,
+    "order": 3,
+}
+
+
+def _sampler_setting_field(knob: object) -> str:
+    knob_name = str(knob)
+    return "sampler_order" if knob_name == "order" else f"sampler_{knob_name}"
+
+
+def _all_settings_for_sampler(sampler_name: str) -> dict:
+    """Build one valid value for every optional setting the SDK exposes."""
+    constraints = SAMPLER_CONSTRAINTS[sampler_name]
+    settings = {_sampler_setting_field(knob): _SAMPLER_SETTING_VALUES[str(knob)] for knob in constraints.numeric_knob_ranges}
+    if constraints.solver_type_choices:
+        settings["sampler_solver_type"] = str(constraints.solver_type_choices[0])
+    settings["scheduler"] = "karras"
+    return settings
+
+
+class SamplerFeatureRequester(HttpUser):
+    """Submit extended samplers with zero, some, or all optional settings."""
+
+    weight = 1
+    fixed_count = 0
     wait_time = between(1, 2)
     max_pending = 8
 
     def on_start(self):
         self.api_key = _pick_requestor_key()
         self.pending_ids = deque()
+        profiles = [(sampler_name, _all_settings_for_sampler(sampler_name), None) for sampler_name in _EXTENDED_IMAGE_SAMPLERS]
+        profiles.append(
+            (
+                "k_euler",
+                {"scheduler": "simple", "flow_shift": 1.1},
+                "Flux.1-Schnell fp8 (Compact)",
+            ),
+        )
+        profiles.extend(
+            ("uni_pc", {"scheduler": scheduler}, None)
+            for scheduler in (
+                "normal",
+                "simple",
+                "sgm_uniform",
+                "exponential",
+                "ddim_uniform",
+                "beta",
+                "linear_quadratic",
+                "kl_optimal",
+            )
+        )
+        profiles.extend(("k_euler", {"scheduler": scheduler}, "stable_diffusion") for scheduler in ("align_your_steps", "gits"))
+        self.feature_cases = cycle(
+            (sampler_name, mode, settings, model)
+            for sampler_name, settings, model in profiles
+            for mode in ("none", "subset", "all")
+            if mode != "subset" or len(settings) > 1
+        )
 
     def _cancel_oldest(self):
         if not self.pending_ids:
@@ -572,6 +670,18 @@ class SamplerFeatureRequester(HttpUser):
         while self.pending_ids:
             self._cancel_oldest()
 
+    def _next_case(self) -> tuple[str, str, dict, str | None]:
+        sampler_name, mode, all_settings, model = next(self.feature_cases)
+        if mode == "none":
+            selected_settings = {}
+        elif mode == "all":
+            selected_settings = all_settings
+        else:
+            count = random.randint(1, len(all_settings) - 1)
+            names = random.sample(list(all_settings), k=count)
+            selected_settings = {name: all_settings[name] for name in names}
+        return sampler_name, mode, selected_settings, model
+
     @tag("image", "cold", "requestor", "sampler-features")
     @task
     def generate_sampler_feature(self):
@@ -580,20 +690,14 @@ class SamplerFeatureRequester(HttpUser):
 
         opts = self.environment.parsed_options
         configured_models = _config.get("models", [])
-        default_feature_models = ["stable_diffusion"] if "stable_diffusion" in configured_models else configured_models[:1]
-
-        # Cases with baseline-specific settings only run when their known model
-        # is in the configured worker pool. Custom deployments can therefore
-        # replace --horde-models without receiving predictable validation 4xxs.
-        case = next(_image_sampler_feature_cases)
-        for _ in range(len(_IMAGE_SAMPLER_FEATURE_CASES)):
-            preferred_models = list(case.get("models", ()))
-            if not preferred_models or all(model in configured_models for model in preferred_models):
-                break
-            case = next(_image_sampler_feature_cases)
-        else:
-            preferred_models = []
-
+        sampler_name, mode, settings, preferred_model = self._next_case()
+        while preferred_model and preferred_model not in configured_models:
+            sampler_name, mode, settings, preferred_model = self._next_case()
+        models = (
+            [preferred_model]
+            if preferred_model
+            else (["stable_diffusion"] if "stable_diffusion" in configured_models else configured_models[:1])
+        )
         payload = {
             "prompt": _random_prompt(),
             "nsfw": False,
@@ -603,17 +707,18 @@ class SamplerFeatureRequester(HttpUser):
                 "width": opts.gen_width,
                 "height": opts.gen_height,
                 "steps": opts.gen_steps,
-                "cfg_scale": 1.5 if case["params"]["sampler_name"].endswith("cfg_pp") else opts.gen_cfg_scale,
-                **case["params"],
+                "cfg_scale": 1.5 if sampler_name.endswith("cfg_pp") else opts.gen_cfg_scale,
+                "sampler_name": sampler_name,
+                **settings,
             },
-            "models": preferred_models or default_feature_models,
+            "models": models,
         }
         with self.client.post(
             "/api/v2/generate/async",
             json=payload,
             headers=_headers(self.api_key),
             catch_response=True,
-            name=f"/api/v2/generate/async [sampler-feature/{case['name']}]",
+            name=f"/api/v2/generate/async [sampler-feature/{mode}]",
         ) as resp:
             req_id = _handle_async_generate(resp, self.environment)
             if req_id:

tests/stress/locustsuite/users/meta.py

@@ -54,7 +54,7 @@ class MetaBrowser(HttpUser):
         # /status/models is @cache.cached, this is the canonical hot path.
         self.client.get("/api/v2/status/models?type=image", name="/api/v2/status/models [hot]")
 
-    @tag("meta", "image", "sampler-features")
+    @tag("meta", "image")
     @task(2)
     def sampler_constraints(self):
         """Published client contract for sampler settings and pricing."""

tests/stress/locustsuite/users/misuse.py

@@ -5,28 +5,12 @@
 """Validation and defensive-path Locust users."""
 
 import uuid
-from itertools import cycle
 
 from locust import HttpUser, between, tag, task
 
 from ..config import _config
 from ..helpers import _headers, _pick_requestor_key, _pick_worker_key, _random_prompt, _record_expected
 
-_invalid_sampler_constraint_cases = cycle(
-    (
-        {"params": {"sampler_name": "k_euler", "scheduler": "karras", "sampler_eta": 1.0}},
-        {"params": {"sampler_name": "deis", "scheduler": "karras", "sampler_order": 1}},
-        {
-            "params": {"sampler_name": "dpmpp_2m_sde", "scheduler": "karras", "sampler_solver_type": "phi_1"},
-        },
-        {"params": {"sampler_name": "dpmpp_3m_sde", "scheduler": "normal"}},
-        {
-            "params": {"sampler_name": "k_euler", "scheduler": "align_your_steps"},
-            "models": ["Flux.1-Schnell fp8 (Compact)"],
-        },
-    ),
-)
-
 
 class MisuseUser(HttpUser):
     """Common endpoint misuse. Exercises validation & auth rejection paths."""
@@ -116,31 +100,6 @@ class MisuseUser(HttpUser):
         ) as resp:
             self._expect_4xx(resp, "/api/v2/generate/async [misuse-oversized]")
 
-    @tag("misuse", "image", "sampler-features")
-    @task(2)
-    def invalid_sampler_constraints(self):
-        """Exercise each new hard-constraint rejection family without creating work."""
-        case = next(_invalid_sampler_constraint_cases)
-        payload = {
-            "prompt": _random_prompt(),
-            "params": {
-                "width": 512,
-                "height": 512,
-                "steps": 20,
-                "cfg_scale": 7.0,
-                **case["params"],
-            },
-            "models": case.get("models", ["stable_diffusion"]),
-        }
-        with self.client.post(
-            "/api/v2/generate/async",
-            json=payload,
-            headers=_headers(_pick_requestor_key()),
-            catch_response=True,
-            name="/api/v2/generate/async [misuse-sampler-constraint]",
-        ) as resp:
-            self._expect_4xx(resp, "/api/v2/generate/async [misuse-sampler-constraint]")
-
     @tag("misuse", "image")
     @task(2)
     def invalid_model_name(self):

uv.lock

@@ -58,6 +58,7 @@ dev = [
     { name = "pytest-cov" },
     { name = "reuse" },
     { name = "ruff" },
+    { name = "sqlparse" },
     { name = "testcontainers", extra = ["redis"] },
     { name = "tox" },
     { name = "types-python-dateutil" },
@@ -120,6 +121,7 @@ dev = [
     { name = "pytest-cov" },
     { name = "reuse" },
     { name = "ruff", specifier = "==0.15.12" },
+    { name = "sqlparse" },
     { name = "testcontainers", extras = ["postgres", "redis"] },
     { name = "tox", specifier = "~=4.14.1" },
     { name = "types-python-dateutil", specifier = ">=2.9.0.20260716" },
@@ -2939,6 +2941,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/e2/22/dbf013a12ec759e54a34a119e9e217435b3f71b2dd5c61a7ade0a25dae87/sqlalchemy-2.0.51-py3-none-any.whl", hash = "sha256:bb024d8b621d0be75f4f44ecc7c950450026e76d66dc8f791bb5331d7fed59d5", size = 1944334, upload-time = "2026-06-15T16:09:22.418Z" },
 ]
 
+[[package]]
+name = "sqlparse"
+version = "0.6.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/5f/d3/3f06a1006f2261d1342aefb3c71eed02f5d4ca5bdbecd86ebc12ad38306e/sqlparse-0.6.0.tar.gz", hash = "sha256:113c35c75365ab9cc9c7231d68c6428fb11c085fc8e9eb1ad659b7ddbf6cd2b9", size = 178477, upload-time = "2026-08-13T19:16:06.396Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/d9/50/f00935da0ec7cbf325f8dc4f772ae46fbc7b672dd62876e73f0a94adda57/sqlparse-0.6.0-py3-none-any.whl", hash = "sha256:b861c0288ce2fa56209a9a6412d2e066ac664b3873b89c26c9d8415e8e32996f", size = 50070, upload-time = "2026-08-13T19:16:04.062Z" },
+]
+
 [[package]]
 name = "strenum"
 version = "0.4.15"