Add Stop-A ON config (adaptive_stop enabled + boundary guard) for A/B

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

configs/examples/dash0_qwen30b_a3b_stopA_on.json

@@ -0,0 +1,147 @@
+{
+  "study_id": "dash0-qwen30b-a3b-stopA-on-chat-0-8k",
+  "hardware": {
+    "gpu_count": 8,
+    "gpu_model": "H20",
+    "host_candidates": [
+      "dash0"
+    ]
+  },
+  "model": {
+    "model_id": "Qwen/Qwen3-30B-A3B",
+    "served_model_name": "qwen3-30b-a3b-community"
+  },
+  "engine": {
+    "engine_name": "vllm",
+    "engine_version": "0.20.0",
+    "exec_path": "/usr/local/bin/vllm",
+    "cwd": "/home/admin/cpfs/wjh/aituner/aituner",
+    "host": "127.0.0.1",
+    "port": 18230,
+    "healthcheck_path": "/v1/models",
+    "ready_timeout_s": 900,
+    "request_timeout_s": 900,
+    "launch_args": [
+      "serve",
+      "/home/admin/cpfs/wjh/models/Qwen/Qwen3-30B-A3B"
+    ],
+    "base_envs": {
+      "CUDA_VISIBLE_DEVICES": "0,1,2,3,4,5,6,7",
+      "HOME": "/tmp/wjh",
+      "XDG_CACHE_HOME": "/tmp/wjh/.cache"
+    },
+    "base_flags": {
+      "host": "127.0.0.1",
+      "port": 18230,
+      "served-model-name": "qwen3-30b-a3b-community",
+      "gpu-memory-utilization": 0.9,
+      "max-model-len": 16384,
+      "trust-remote-code": true,
+      "enable-prefix-caching": true
+    },
+    "tunable_envs": [],
+    "tunable_flags": [
+      "tensor-parallel-size",
+      "data-parallel-size",
+      "enable-expert-parallel",
+      "expert-parallel-size",
+      "gpu-memory-utilization",
+      "max-num-batched-tokens",
+      "max-num-seqs",
+      "block-size",
+      "enable-prefix-caching",
+      "enable-chunked-prefill"
+    ],
+    "topology_constraints": {
+      "require_tp_dp_product_equals_gpu_count": false,
+      "require_ep_size_leq_tp_dp_product": true,
+      "require_ep_size_divides_tp_dp_product": true,
+      "require_enable_expert_parallel_when_ep_gt_one": true,
+      "validate_cuda_graph_sizes_divisible_by_tp_when_tp_ep_reduce_scatter": true,
+      "allowed_tp_dp_products": [
+        1,
+        2,
+        4,
+        8
+      ],
+      "allowed_tensor_parallel_sizes": [
+        1,
+        2,
+        4,
+        8
+      ],
+      "allowed_data_parallel_sizes": [
+        1,
+        2,
+        4,
+        8
+      ],
+      "allowed_expert_parallel_sizes": [
+        1,
+        2,
+        4,
+        8
+      ]
+    },
+    "python_executable": "/tmp/wjh/venvs/vllm-0.20.0-cu129/bin/python"
+  },
+  "trace": {
+    "windows_path": "/home/admin/cpfs/wjh/aituner/aituner/trace_windows/windows.json",
+    "window_id": "chat_w20260311_1000",
+    "completion_tokens_override": 128,
+    "u_field": "sampling_u",
+    "timestamp_field": "timestamp",
+    "max_concurrency": 64,
+    "input_length_filter": {
+      "min_input_tokens": 0,
+      "max_input_tokens": 8192
+    },
+    "replay_time_scale": 1.0,
+    "early_stop_max_lag_s": 120.0,
+    "early_stop_max_elapsed_s": 900.0,
+    "adaptive_stop": {
+      "enabled": true,
+      "tau": 0.9,
+      "tau_c": 0.9,
+      "stable_checks": 3,
+      "max_checks": 20,
+      "min_fraction": 0.1,
+      "boundary_delta": 0.02
+    }
+  },
+  "slo": {
+    "target_pass_rate": 0.95,
+    "ttft_rule": {
+      "kind": "step_ms",
+      "buckets": [
+        {
+          "max_input_tokens": 4096,
+          "threshold_ms": 2000
+        },
+        {
+          "max_input_tokens": 32768,
+          "threshold_ms": 4000
+        },
+        {
+          "threshold_ms": 6000
+        }
+      ]
+    },
+    "tpot_rule": {
+      "kind": "fixed_ms",
+      "threshold_ms": 50
+    }
+  },
+  "search": {
+    "low": 0.0,
+    "high": 0.125,
+    "tolerance": 0.001,
+    "max_probes": 4,
+    "sample_seed": 20260325
+  },
+  "llm": {
+    "system_prompt": "Tune community vLLM 0.20.0 serving for Qwen3-30B-A3B. Start from the default vLLM engine configuration, use only launch-safe patches, and optimize request_rate_per_gpu under the configured SLO.",
+    "max_history_trials": 8,
+    "use_harness": false
+  }
+}

src/aituner/spec.py

@@ -335,6 +335,7 @@ class AdaptiveStopSpec:
     stable_checks: int = 3
     max_checks: int = 20
     min_fraction: float = 0.1
+    boundary_delta: float = 0.02
 
     @classmethod
     def from_dict(cls, data: Any) -> "AdaptiveStopSpec":
@@ -357,9 +358,14 @@ class AdaptiveStopSpec:
         min_fraction = _require_float(
             m.get("min_fraction", 0.1), context="trace.adaptive_stop.min_fraction"
         )
+        boundary_delta = _require_float(
+            m.get("boundary_delta", 0.02), context="trace.adaptive_stop.boundary_delta"
+        )
         for name, value in (("tau", tau), ("tau_c", tau_c), ("min_fraction", min_fraction)):
             if not 0.0 < value <= 1.0:
                 raise SpecError(f"trace.adaptive_stop.{name} must be in (0, 1].")
+        if not 0.0 <= boundary_delta < 1.0:
+            raise SpecError("trace.adaptive_stop.boundary_delta must be in [0, 1).")
         if stable_checks <= 0 or max_checks <= 0:
             raise SpecError(
                 "trace.adaptive_stop.stable_checks and max_checks must be > 0."
@@ -376,6 +382,7 @@ class AdaptiveStopSpec:
             stable_checks=stable_checks,
             max_checks=max_checks,
             min_fraction=min_fraction,
+            boundary_delta=boundary_delta,
         )
 
 

src/aituner/worker.py

@@ -249,6 +249,29 @@ def _adaptive_replay_set(
     return replay, certificate
 
 
+def _should_extend_on_boundary(
+    *,
+    pass_rate: float,
+    target_pass_rate: float,
+    certificate: dict[str, Any] | None,
+    truncated: bool,
+    boundary_delta: float,
+) -> bool:
+    """SLO-boundary guard: re-measure on the full window when a truncated probe
+    lands within +/- boundary_delta of the SLO target.
+
+    Offered-L-C-A convergence cannot see engine-state drift in the window's tail,
+    so a near-boundary truncated verdict is untrustworthy. This fires only on
+    probes sitting on the feasibility knee, so non-boundary probes keep the Stop-A
+    time saving.
+    """
+    if certificate is None or not certificate.get("converged"):
+        return False
+    if not truncated or boundary_delta <= 0:
+        return False
+    return abs(float(pass_rate) - float(target_pass_rate)) <= float(boundary_delta)
+
+
 def _best_feasible_probe_record(probe_history: list[dict[str, Any]]) -> dict[str, Any] | None:
     feasible = [
         item
@@ -563,18 +586,36 @@ def run_trial(trial_spec_path: Path) -> dict[str, Any]:
                     selected, study=study, window=window
                 )
                 restart_after_early_stop = study.trace.restart_engine_after_early_stop
-                outcomes, early_stopped, early_stop_reason = _replay_requests(
+
-                    replay_set,
+                def _run(reqs: list[TraceRequest]) -> tuple[list[RequestOutcome], bool, str]:
-                    base_url=recipe.base_url,
+                    return _replay_requests(
-                    timeout_s=recipe.request_timeout_s,
+                        reqs,
-                    max_concurrency=study.trace.max_concurrency,
+                        base_url=recipe.base_url,
-                    target_pass_rate=study.slo.target_pass_rate,
+                        timeout_s=recipe.request_timeout_s,
-                    max_lag_s=study.trace.early_stop_max_lag_s,
+                        max_concurrency=study.trace.max_concurrency,
-                    max_elapsed_s=study.trace.early_stop_max_elapsed_s,
+                        target_pass_rate=study.slo.target_pass_rate,
-                    evaluate_outcome=lambda outcome: evaluate_request(outcome, study.slo),
+                        max_lag_s=study.trace.early_stop_max_lag_s,
-                    drain_inflight_on_early_stop=not restart_after_early_stop,
+                        max_elapsed_s=study.trace.early_stop_max_elapsed_s,
-                )
+                        evaluate_outcome=lambda outcome: evaluate_request(outcome, study.slo),
+                        drain_inflight_on_early_stop=not restart_after_early_stop,
+                    )
+
+                outcomes, early_stopped, early_stop_reason = _run(replay_set)
                 evaluations, summary = summarize_evaluations(outcomes, study.slo)
+                if _should_extend_on_boundary(
+                    pass_rate=summary["slo_pass_rate"],
+                    target_pass_rate=study.slo.target_pass_rate,
+                    certificate=adaptive_stop_certificate,
+                    truncated=len(replay_set) < len(selected),
+                    boundary_delta=study.trace.adaptive_stop.boundary_delta,
+                ):
+                    # On the feasibility knee the truncated verdict is untrustworthy;
+                    # re-measure the full window and use that result.
+                    replay_set = selected
+                    outcomes, early_stopped, early_stop_reason = _run(selected)
+                    evaluations, summary = summarize_evaluations(outcomes, study.slo)
+                    if adaptive_stop_certificate is not None:
+                        adaptive_stop_certificate["boundary_extended"] = True
                 probe_details = _probe_outcome_details(
                     threshold=threshold,
                     selected=replay_set,

tests/test_core_flow.py

@@ -53,6 +53,7 @@ from aituner.store import StudyStore
 from aituner.trace import load_trace_requests, summarize_window
 from aituner.worker import (
     _adaptive_replay_set,
+    _should_extend_on_boundary,
     _best_feasible_probe_record,
     _latency_summary,
     _run_one_request,
@@ -476,6 +477,60 @@ class CoreFlowTests(unittest.TestCase):
         self.assertIsNone(no_cert)
         self.assertEqual(len(passthrough), len(requests))
 
+    def test_boundary_guard_extends_only_near_the_slo_knee(self) -> None:
+        converged = {"converged": True}
+        # Truncated, converged, pass-rate on the knee -> re-measure full.
+        self.assertTrue(
+            _should_extend_on_boundary(
+                pass_rate=0.961, target_pass_rate=0.95, certificate=converged,
+                truncated=True, boundary_delta=0.02,
+            )
+        )
+        self.assertTrue(
+            _should_extend_on_boundary(
+                pass_rate=0.946, target_pass_rate=0.95, certificate=converged,
+                truncated=True, boundary_delta=0.02,
+            )
+        )
+        # Clearly feasible / clearly infeasible -> trust the truncated verdict.
+        self.assertFalse(
+            _should_extend_on_boundary(
+                pass_rate=0.99, target_pass_rate=0.95, certificate=converged,
+                truncated=True, boundary_delta=0.02,
+            )
+        )
+        self.assertFalse(
+            _should_extend_on_boundary(
+                pass_rate=0.50, target_pass_rate=0.95, certificate=converged,
+                truncated=True, boundary_delta=0.02,
+            )
+        )
+        # Not truncated, not converged, guard disabled, or no certificate -> no extend.
+        self.assertFalse(
+            _should_extend_on_boundary(
+                pass_rate=0.95, target_pass_rate=0.95, certificate=converged,
+                truncated=False, boundary_delta=0.02,
+            )
+        )
+        self.assertFalse(
+            _should_extend_on_boundary(
+                pass_rate=0.95, target_pass_rate=0.95, certificate={"converged": False},
+                truncated=True, boundary_delta=0.02,
+            )
+        )
+        self.assertFalse(
+            _should_extend_on_boundary(
+                pass_rate=0.95, target_pass_rate=0.95, certificate=converged,
+                truncated=True, boundary_delta=0.0,
+            )
+        )
+        self.assertFalse(
+            _should_extend_on_boundary(
+                pass_rate=0.95, target_pass_rate=0.95, certificate=None,
+                truncated=True, boundary_delta=0.02,
+            )
+        )
+
     def test_lca_similarity_matrix_separates_different_profiles(self) -> None:
         window = WindowRecord(
             window_id="base",