Add consolidated two-stop summary doc

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

configs/examples/dash0_qwen27b_stopB_loop.json

@@ -0,0 +1,177 @@
+{
+  "study_id": "dash0-qwen27b-stopB-loop-chat-0-8k",
+  "hardware": {
+    "gpu_count": 8,
+    "gpu_model": "H20",
+    "host_candidates": [
+      "dash0"
+    ]
+  },
+  "model": {
+    "model_id": "qwen3.5-27b-256k-0223-internal",
+    "served_model_name": "qwen35-27b-aituner"
+  },
+  "engine": {
+    "engine_name": "vllm",
+    "engine_version": "latest-release-on-dash0",
+    "exec_path": "/usr/local/bin/vllm",
+    "cwd": "/home/admin/cpfs/wjh/aituner/aituner",
+    "host": "127.0.0.1",
+    "port": 18082,
+    "healthcheck_path": "/v1/models",
+    "ready_timeout_s": 900,
+    "request_timeout_s": 180,
+    "launch_args": [
+      "serve",
+      "/home/admin/resource/model/464482ce/qwen3.5-27b/256k-0223-internal"
+    ],
+    "base_envs": {
+      "VLLM_DISABLE_COMPILE_CACHE": "1",
+      "DS_LLM_IGNORE_WARMUP": "1",
+      "DS_LLM_IGNORE_CHECK_WARMUP": "1",
+      "VLLM_ENABLE_MODEL_RUNNER_WARMUP": "1",
+      "VLLM_GDN_USE_FUSED_QKVZBA_KERNEL": "0",
+      "PARAM_TOTAL_MAX": "262144",
+      "PARAM_IN_LENGTH_MAX": "262144",
+      "PARAM_MAX_LENGTH_MAX": "131072",
+      "DS_LLM_MAX_THINK_TOKENS": "81920",
+      "DS_LLM_GRACEFUL_SHUTDOWN_WAIT_SECONDS": "600",
+      "VLLM_FP8_USE_BLADNN": "1",
+      "VLLM_MOE_USE_BLADNN": "1",
+      "VLLM_GDN_USE_BLADNN": "0",
+      "VLLM_USE_V1": "1",
+      "VLLM_IS_HYBRID_MODEL": "1",
+      "VLLM_ENABLE_TORCH_COMPILE": "1",
+      "VLLM_ATTENTION_BACKEND": "FLASH_ATTN",
+      "VLLM_QUANTIZE_ROUTED_EXPERTS_ONLY": "1",
+      "VLLM_USE_FLASHINFER_SAMPLER": "0",
+      "VLLM_DP_MASTER_PORT": "9528",
+      "VLLM_RESPONSE_TIMEOUT": "300",
+      "VLLM_LOG_REQ_KV_LENS": "1",
+      "DS_LLM_GRACEFUL_SHUTDOWN_KEEP_SECONDS": "600",
+      "CUDA_VISIBLE_DEVICES": "2,3,4,5,6,7"
+    },
+    "base_flags": {
+      "host": "127.0.0.1",
+      "port": 18082,
+      "served-model-name": "qwen35-27b-aituner",
+      "trust-remote-code": true,
+      "dtype": "bfloat16",
+      "gpu-memory-utilization": 0.9,
+      "enable-prefix-caching": true,
+      "mamba-cache-mode": "light",
+      "distributed-executor-backend": "mp",
+      "block-size": 64,
+      "enable-chunked-prefill": true,
+      "max-num-batched-tokens": 8192,
+      "disable-cascade-attn": true,
+      "max-model-len": 262144,
+      "speculative-config": "{\"method\":\"qwen3_next_vl_mtp\",\"num_speculative_tokens\":3}",
+      "mm-processor-cache-gb": 0,
+      "limit-mm-per-prompt": "{\"image\":256,\"video\":64}",
+      "compilation-config": "{\"cudagraph_mode\":\"FULL_AND_PIECEWISE\",\"use_inductor\":false,\"pass_config\":{\"fuse_norm_quant\":false,\"fuse_act_quant\":false,\"fuse_attn_quant\":false}}",
+      "mamba-cache-dtype": "float32",
+      "skip-mm-profiling": true,
+      "quantization": "fp8",
+      "tensor-parallel-size": 1,
+      "disable-log-requests": true
+    },
+    "tunable_envs": [
+      "VLLM_ENABLE_TORCH_COMPILE"
+    ],
+    "tunable_flags": [
+      "tensor-parallel-size",
+      "data-parallel-size",
+      "expert-parallel-size",
+      "gpu-memory-utilization",
+      "block-size",
+      "max-num-batched-tokens",
+      "max-num-seqs",
+      "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
+      ]
+    },
+    "python_executable": "python3"
+  },
+  "trace": {
+    "windows_path": "/home/admin/cpfs/wjh/aituner/aituner/trace_windows/windows.json",
+    "window_id": "chat_w20260311_1000",
+    "u_field": "sampling_u",
+    "timestamp_field": "timestamp",
+    "max_concurrency": 32,
+    "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": "linear_ms",
+      "intercept_ms": 4000,
+      "per_token_ms": 0.125
+    },
+    "tpot_rule": {
+      "kind": "fixed_ms",
+      "threshold_ms": 50
+    }
+  },
+  "search": {
+    "low": 0.0,
+    "high": 0.25,
+    "tolerance": 0.001,
+    "max_probes": 6,
+    "sample_seed": 20260325,
+    "inherit_incumbent_floor": true
+  },
+  "llm": {
+    "system_prompt": "Propose a single engine config patch that increases the maximum feasible sampling_u under the SLO target. Favor launch-safe changes grounded in the incumbent result and only propose knobs that plausibly improve throughput above the incumbent request rate.",
+    "max_history_trials": 8,
+    "endpoint": {
+      "provider": "codex",
+      "model": "gpt-5.4",
+      "stream": true,
+      "api_key_env": "OPENAI_API_KEY",
+      "timeout_s": 180
+    }
+  }
+}

configs/examples/dash0_qwen27b_tp_ab.json

@@ -0,0 +1,176 @@
+{
+  "study_id": "dash0-qwen27b-tp-ab-chat-0-8k",
+  "hardware": {
+    "gpu_count": 8,
+    "gpu_model": "H20",
+    "host_candidates": [
+      "dash0"
+    ]
+  },
+  "model": {
+    "model_id": "qwen3.5-27b-256k-0223-internal",
+    "served_model_name": "qwen35-27b-aituner"
+  },
+  "engine": {
+    "engine_name": "vllm",
+    "engine_version": "latest-release-on-dash0",
+    "exec_path": "/usr/local/bin/vllm",
+    "cwd": "/home/admin/cpfs/wjh/aituner/aituner",
+    "host": "127.0.0.1",
+    "port": 18082,
+    "healthcheck_path": "/v1/models",
+    "ready_timeout_s": 900,
+    "request_timeout_s": 180,
+    "launch_args": [
+      "serve",
+      "/home/admin/resource/model/464482ce/qwen3.5-27b/256k-0223-internal"
+    ],
+    "base_envs": {
+      "VLLM_DISABLE_COMPILE_CACHE": "1",
+      "DS_LLM_IGNORE_WARMUP": "1",
+      "DS_LLM_IGNORE_CHECK_WARMUP": "1",
+      "VLLM_ENABLE_MODEL_RUNNER_WARMUP": "1",
+      "VLLM_GDN_USE_FUSED_QKVZBA_KERNEL": "0",
+      "PARAM_TOTAL_MAX": "262144",
+      "PARAM_IN_LENGTH_MAX": "262144",
+      "PARAM_MAX_LENGTH_MAX": "131072",
+      "DS_LLM_MAX_THINK_TOKENS": "81920",
+      "DS_LLM_GRACEFUL_SHUTDOWN_WAIT_SECONDS": "600",
+      "VLLM_FP8_USE_BLADNN": "1",
+      "VLLM_MOE_USE_BLADNN": "1",
+      "VLLM_GDN_USE_BLADNN": "0",
+      "VLLM_USE_V1": "1",
+      "VLLM_IS_HYBRID_MODEL": "1",
+      "VLLM_ENABLE_TORCH_COMPILE": "1",
+      "VLLM_ATTENTION_BACKEND": "FLASH_ATTN",
+      "VLLM_QUANTIZE_ROUTED_EXPERTS_ONLY": "1",
+      "VLLM_USE_FLASHINFER_SAMPLER": "0",
+      "VLLM_DP_MASTER_PORT": "9528",
+      "VLLM_RESPONSE_TIMEOUT": "300",
+      "VLLM_LOG_REQ_KV_LENS": "1",
+      "DS_LLM_GRACEFUL_SHUTDOWN_KEEP_SECONDS": "600",
+      "CUDA_VISIBLE_DEVICES": "2,3,4,5,6,7"
+    },
+    "base_flags": {
+      "host": "127.0.0.1",
+      "port": 18082,
+      "served-model-name": "qwen35-27b-aituner",
+      "trust-remote-code": true,
+      "dtype": "bfloat16",
+      "gpu-memory-utilization": 0.9,
+      "enable-prefix-caching": true,
+      "mamba-cache-mode": "light",
+      "distributed-executor-backend": "mp",
+      "block-size": 64,
+      "enable-chunked-prefill": true,
+      "max-num-batched-tokens": 8192,
+      "disable-cascade-attn": true,
+      "max-model-len": 262144,
+      "speculative-config": "{\"method\":\"qwen3_next_vl_mtp\",\"num_speculative_tokens\":3}",
+      "mm-processor-cache-gb": 0,
+      "limit-mm-per-prompt": "{\"image\":256,\"video\":64}",
+      "compilation-config": "{\"cudagraph_mode\":\"FULL_AND_PIECEWISE\",\"use_inductor\":false,\"pass_config\":{\"fuse_norm_quant\":false,\"fuse_act_quant\":false,\"fuse_attn_quant\":false}}",
+      "mamba-cache-dtype": "float32",
+      "skip-mm-profiling": true,
+      "quantization": "fp8",
+      "tensor-parallel-size": 1,
+      "disable-log-requests": true
+    },
+    "tunable_envs": [
+      "VLLM_ENABLE_TORCH_COMPILE"
+    ],
+    "tunable_flags": [
+      "tensor-parallel-size",
+      "data-parallel-size",
+      "expert-parallel-size",
+      "gpu-memory-utilization",
+      "block-size",
+      "max-num-batched-tokens",
+      "max-num-seqs",
+      "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
+      ]
+    },
+    "python_executable": "python3"
+  },
+  "trace": {
+    "windows_path": "/home/admin/cpfs/wjh/aituner/aituner/trace_windows/windows.json",
+    "window_id": "chat_w20260311_1000",
+    "u_field": "sampling_u",
+    "timestamp_field": "timestamp",
+    "max_concurrency": 32,
+    "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": "linear_ms",
+      "intercept_ms": 4000,
+      "per_token_ms": 0.125
+    },
+    "tpot_rule": {
+      "kind": "fixed_ms",
+      "threshold_ms": 50
+    }
+  },
+  "search": {
+    "low": 0.0,
+    "high": 0.125,
+    "tolerance": 0.001,
+    "max_probes": 6,
+    "sample_seed": 20260325
+  },
+  "llm": {
+    "system_prompt": "Propose a single engine config patch that increases the maximum feasible sampling_u under the SLO target. Favor launch-safe changes grounded in the incumbent result and only propose knobs that plausibly improve throughput above the incumbent request rate.",
+    "max_history_trials": 8,
+    "endpoint": {
+      "provider": "codex",
+      "model": "gpt-5.4",
+      "stream": true,
+      "api_key_env": "OPENAI_API_KEY",
+      "timeout_s": 180
+    }
+  }
+}

configs/examples/dash0_qwen30b_a3b_stopA_fulldata.json

@@ -0,0 +1,138 @@
+{
+  "study_id": "dash0-qwen30b-a3b-stopA-fulldata-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
+  },
+  "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
+  }
+}

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
+  }
+}

configs/examples/dash0_qwen30b_a3b_stopB_e2e.json

@@ -0,0 +1,155 @@
+{
+  "study_id": "dash0-qwen30b-a3b-stopB-e2e-hi-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
+    },
+    "max_requests_per_probe": 512,
+    "replay_time_scale": 0.1,
+    "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": 1.0,
+    "tolerance": 0.001,
+    "max_probes": 6,
+    "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": true,
+    "endpoint": {
+      "provider": "codex",
+      "model": "gpt-5.4",
+      "stream": true,
+      "api_key_env": "OPENAI_API_KEY",
+      "timeout_s": 240
+    }
+  }
+}

configs/examples/stopb_27b_ab/p1_tp1.json

@@ -0,0 +1,13 @@
+{
+  "observation": "baseline TP1 (deployed flags)",
+  "diagnosis": "deterministic TP A/B point",
+  "config_patch": {
+    "env_patch": {},
+    "flag_patch": {}
+  },
+  "expected_effects": [
+    "measure peak request_rate_per_gpu at this topology"
+  ],
+  "why_not_previous_failures": "n/a",
+  "should_stop": false
+}

configs/examples/stopb_27b_ab/p2_tp2.json

@@ -0,0 +1,15 @@
+{
+  "observation": "TP2",
+  "diagnosis": "deterministic TP A/B point",
+  "config_patch": {
+    "env_patch": {},
+    "flag_patch": {
+      "tensor-parallel-size": 2
+    }
+  },
+  "expected_effects": [
+    "measure peak request_rate_per_gpu at this topology"
+  ],
+  "why_not_previous_failures": "n/a",
+  "should_stop": false
+}

configs/examples/stopb_27b_ab/p3_tp4.json

@@ -0,0 +1,15 @@
+{
+  "observation": "TP4",
+  "diagnosis": "deterministic TP A/B point",
+  "config_patch": {
+    "env_patch": {},
+    "flag_patch": {
+      "tensor-parallel-size": 4
+    }
+  },
+  "expected_effects": [
+    "measure peak request_rate_per_gpu at this topology"
+  ],
+  "why_not_previous_failures": "n/a",
+  "should_stop": false
+}

docs/harness-ablation/qwen27b-tp-sweep-20260616.md

@@ -0,0 +1,51 @@
+# Qwen3.5-27B TP sweep under length-aware TTFT SLO — 2026-06-16
+
+Branch `feat/two-stop`. Deterministic ground-truth A/B (proposal files, no LLM):
+TP1 vs TP2 vs TP4 on the dense Qwen3.5-27B (internal 256k, fp8, spec-decode) at
+0–8k chat, vLLM 0.11.1, H20, `replay_time_scale=1.0` (no smoke), Stop-A enabled,
+pinned to GPUs 2–7.
+
+**SLO**: TTFT ≤ `4000 + 0.125·L_in` ms (= 4s + L_in/8k), TPOT ≤ 50 ms, pass ≥ 95%.
+
+## Result
+
+| config | best_u | raw req/s | req/s/GPU | pass | saturated |
+| --- | --- | --- | --- | --- | --- |
+| TP1 | 0.00195 | 0.065 | **0.065** | 1.00 | no |
+| TP2 | 0.0195 | 0.585 | **0.2925** | 0.96 | no |
+| TP4 | 0.123 | 3.63 | **≥0.908** | 0.98 | **yes (best_u≈high=0.125)** |
+
+- **Per-GPU throughput rises sharply with TP for the dense 27B**: TP2 = 4.5× TP1,
+  TP4 ≥ 14× TP1. Opposite of the MoE Qwen3-30B-A3B (TP1 best per-GPU) — confirms the
+  dense-vs-MoE distinction.
+- **Mechanism**: TP1 is TPOT-bound — one H20 cannot decode a 27B under 50 ms/token
+  once the batch grows, so it saturates at ~0.065 req/s/GPU. Loosening TTFT (2s→4-5s)
+  did *not* change TP1 (still 0.065), confirming TPOT — not TTFT — is TP1's binding
+  constraint. Each TP doubling speeds decode+prefill enough to more than recover the
+  added GPUs.
+- **TP4 saturated** the offered-load ceiling (`best_u=0.123 ≈ 0.125`): still feasible
+  after ~the whole trace, so 0.908 is a lower bound. True peak (and TP8) need a
+  raised `search.high` to measure.
+
+## Process findings (fed back into the harness)
+
+- **Bug fixed**: a request exceeding `request_timeout_s` raised a raw `TimeoutError`
+  mid-stream that escaped `_run_one_request` and crashed the whole trial; now wrapped
+  as `HttpClientError` (failed request, not failed trial). Commit `2fcaf80`.
+- **Open gap**: killing a `study tune` run orphans the `VLLM::EngineCore` workers
+  (SIGTERM/SIGKILL of the loop doesn't tear down the engine), which twice left leaked
+  GPU memory on GPUs 0/1 (dead PIDs still pinning KV, only clearable via root
+  `nvidia-smi --gpu-reset`). Fix: SIGTERM handler in the CLI loop + make
+  `_terminate_process_tree` match `EngineCore` workers, not just `vllm serve`.
+- Experiment hygiene: scale=1.0 makes each probe take real arrival time; `search.high`
+  must bracket the config's boundary (too wide wastes probes on a low-capacity config;
+  too low saturates a high-capacity one), and `request_timeout_s` must be modest so
+  overloaded probes drain fast.
+
+## Next
+
+- Re-measure TP4 (and TP8) with `search.high` raised (e.g. 0.5) to find the true peak
+  per-GPU and the TP knee.
+- Run the Stop-B agentic loop on this 27B stack: unlike the 30B (baseline already
+  optimal), here the loop should climb TP1→TP2→TP4 and stop — a real improving
+  trajectory (the original Phase-5 "A" goal).

docs/harness-ablation/stop-a-validation-20260615.md

@@ -0,0 +1,121 @@
+# Stop-A validation (Phase 3) — 2026-06-15
+
+Branch `feat/two-stop`. Stop-A = truncate each probe's replay once the offered
+L-C-A of the replayed prefix converges to the full set (pure L-C-A criterion +
+C-gate). This note records the CPU calibration and the GPU fidelity check.
+
+## 1. Calibration (CPU, no serving)
+
+`scripts/stop_a_calibration.py` on the dash0 0321 10:00–10:10 windows:
+
+| dim | chat (19239 req, hit≈7%) | coder (2451 req, structured reuse) |
+| --- | --- | --- |
+| A | ≥0.95 by frac 0.10 | fast |
+| L | ≥0.96 from frac 0.05 | 0.05=0.75 (heavy tail) → ≥0.94 by 0.20 |
+| **C (slowest)** | noisy, dips (0.50→0.885, 0.55→0.835), stable ≥0.92 only ~0.85 | smooth, stable ≥0.92 by ~0.70 |
+
+Stop fraction (τ_L=τ_A=0.90, W=3):
+
+| τ_c | chat | coder |
+| --- | --- | --- |
+| 0.85 | 0.45 (273s) | 0.45 (255s) |
+| 0.90 | 0.70 (423s) | 0.55 (318s) |
+| 0.92 | 0.85 (513s) | 0.70 (411s) |
+
+Findings:
+- **C is the slowest dimension in both workloads** — reproduces paper §5.2 / Fig 9.
+- **What makes C hard to call converged is signal *noise*, not reuse magnitude.**
+  Low-reuse chat has a sparse/spiky ideal-hit-length series, so its C similarity
+  oscillates and is *harder* to stabilize than the structured, higher-reuse coder.
+  Consequence: a strict τ_c (0.92) gives chat only ~15% saving. A more robust C
+  feature for the low-reuse regime is future work.
+
+## 2. GPU fidelity check (Qwen3-30B-A3B, vLLM 0.11.1, H20)
+
+One full-window run (`adaptive_stop` disabled, `replay_time_scale=1.0`, window
+`chat_w20260311_1000`, 0–8k, out=128), then `scripts/stop_a_validate.py`
+recomputes each probe's convergence prefix and compares the truncated verdict to
+the full verdict — so a single GPU run validates truncation fidelity (no second run).
+
+Trial result: best feasible `sampling_u=0.078125`, request_rate **2.30 req/s**,
+pass_rate 0.973.
+
+Per-probe verdict (τ=0.9):
+
+| τ_c | verdict matches | mean replay saved |
+| --- | --- | --- |
+| 0.85 | 3/4 | 54% |
+| 0.90 | 3/4 | 52% |
+| 0.92 | 3/4 | 38% |
+
+The mismatch is the same probe at every τ_c — the feasibility knee `0.08594`:
+
+```
+thresh    full_pass  prefix_pass  full_feas  prefix_feas
+0.08594   0.946      0.956–0.961  False      True   <- mismatch
+0.07812   0.973      0.987–0.990  True       True
+0.06250   0.986      1.000        True       True
+0.09375   0.268      0.49–0.54    False      False
+```
+
+## 3. Interpretation
+
+- **Stop-A works and saves ~50% of replay** (vs the full 600 s window) while
+  preserving 3/4 probe verdicts. (The paper's ~70% is vs a 30-min fixed baseline;
+  our baseline is the 600 s window, so the percentages are not directly comparable.)
+- **The one failure is a boundary false-positive at the feasibility knee.** At
+  `0.08594` the full window is 0.946 (just below the 0.95 SLO) but the prefix is
+  0.956–0.961 (just above): the *second half* of the window degraded — engine-state
+  drift (KV fill / fragmentation / later-arriving harder requests) that the
+  *offered* L-C-A cannot see. The C-gate did not help because offered-C had
+  converged; the divergence is in the measured pass-rate, not in C.
+- If Stop-A were enabled, the binary search would accept `0.08594`, overestimating
+  the peak sustainable rate by one binary step (~10%).
+
+**This is the boundary jitter we accepted when choosing the pure-L-C-A criterion.**
+The data now argues for revisiting the previously-declined **SLO-boundary guard**:
+keep replaying while the measured pass-rate is within ±δ of the target, even after
+L-C-A converges. It targets exactly this knee case at low extra cost (it only
+extends replay on probes sitting on the feasibility boundary). Recommend adding it
+as a small Stop-A enhancement before enabling Stop-A in production studies.
+
+## 4. SLO-boundary guard (implemented + validated)
+
+Added `trace.adaptive_stop.boundary_delta` (default 0.02): when a truncated probe's
+measured pass-rate lands within ±δ of the SLO target, re-measure on the full window
+and use that verdict. Re-ran the same config with `adaptive_stop` enabled
+(τ=0.9, τ_c=0.90, δ=0.02):
+
+| threshold | feasible | pass | selected | replayed | boundary_extended |
+| --- | --- | --- | --- | --- | --- |
+| 0.06250 | True | 1.000 | 1086 | 487 (45%) | — |
+| 0.09375 | False | 0.444 | 1656 | 822 (50%) | — |
+| 0.07812 | True | 0.994 | 1378 | 682 (49%) | — |
+| 0.08594 | **False** | 0.947 | 1523 | **1523 (100%)** | **True** |
+
+Result: best feasible `sampling_u=0.078125` (rate 2.30 req/s) — **identical to the
+full-replay baseline**. The guard fired on exactly the one knee probe and
+re-measured it to the correct infeasible verdict; the other three probes truncated
+to ~45–50%. Net replayed 3514/5643 requests ≈ **38% replay saved on this trial
+while recovering the correct peak rate** (no one-step overestimate).
+
+**Conclusion: Stop-A with the boundary guard is correct (verdict matches full
+replay) and still saves replay time. Safe to enable.** Configs:
+`dash0_qwen30b_a3b_stopA_fulldata.json` (OFF baseline) and
+`dash0_qwen30b_a3b_stopA_on.json` (ON).
+
+## Repro
+
+```
+# calibration
+PYTHONPATH=src python3 scripts/stop_a_calibration.py \
+  --jsonl <DIR>/qwen_chat_blksz_64_032109-032111.jsonl --block-size 64 \
+  --window-start 3600 --window-end 4200 --gpu-count 8 --label chat
+# GPU run + fidelity
+PYTHONPATH=src python3 -m aituner.cli study tune \
+  --spec configs/examples/dash0_qwen30b_a3b_stopA_fulldata.json \
+  --store-root .aituner/stopA-fulldata --max-trials 1
+PYTHONPATH=src python3 scripts/stop_a_validate.py \
+  --spec configs/examples/dash0_qwen30b_a3b_stopA_fulldata.json \
+  --store-root .aituner/stopA-fulldata --tau 0.9 --tau-c 0.90
+```

docs/harness-ablation/stop-b-e2e-20260615.md

@@ -0,0 +1,86 @@
+# Stop-B end-to-end validation (Phase 5) — 2026-06-15
+
+Branch `feat/two-stop`. Real agentic loop on dash0: Qwen3-30B-A3B / vLLM 0.11.1 /
+8×H20, `gpt-5.4` (via codex/prism) proposing configs, Stop-A enabled to accelerate
+each evaluation, `use_harness=True` so the Stop-B deterministic validator + LLM-stop
+veto are active. Config `dash0_qwen30b_a3b_stopB_e2e.json`, `search.high=1.0`,
+`max_probes=6`, `--max-trials 8`.
+
+## Two stop paths exercised
+
+**Run A (`search.high=0.125`)** — the default config already saturates the offered-load
+search range, so Stop-B fired immediately via the **search-high-saturation** path:
+`stop_authorized_by: validator`, reason *"the incumbent's highest measured probe is
+feasible and within the binary-search resolution of search.high."* Correct
+measurement-ceiling stop (no point proposing configs when the load range, not the
+config, is the bound).
+
+**Run B (`search.high=1.0`)** — forces a real multi-iteration search:
+
+| trial | TP | DP | EP | feasible | raw req/s | **req/s/GPU** | source |
+| --- | --- | --- | --- | --- | --- | --- | --- |
+| 0001 | 1 | 1 | 1 | ✓ | 2.90 | **2.900** | baseline |
+| 0002 | 2 | 1 | 1 | ✓ | 4.42 | 2.208 | harness TP-seed |
+| 0003 | 2 | 1 | 2 | ✗ launch-fail | — | — | harness (EP) |
+| 0004 | 1 | 2 | 1 | ✓ | 4.42 | 2.208 | LLM (after veto) |
+| 0005 | 2 | 2 | 1 | ✓ | 8.37 | 2.092 | harness |
+| 0006 | 2 | 2 | 2 | ✗ launch-fail | — | — | harness (EP) |
+| 0007 | 4 | 1 | 1 | ✓ | 8.37 | 2.092 | LLM |
+| (0008) | — | — | — | **STOP** | — | — | LLM stop, honored after veto budget |
+
+Incumbent: **trial-0001 (TP1), 2.90 req/s/GPU — never beaten.**
+
+> **⚠️ The per-GPU trajectory above is NOT a valid benchmark — it validates only
+> the Stop-B *mechanics*.** Two confounds:
+> 1. **Trace-ceiling saturation.** TP2·DP2 and TP4 reached `best_sampling_u≈0.98`
+>    (still feasible after consuming ~the whole window), so their *true* peak
+>    per-GPU is higher than the 2.09 shown — we ran out of offered load to push
+>    them to their boundary. Only TP1 (u=0.31), TP2 (u=0.48) and DP2 (u=0.48)
+>    found real boundaries. The `sampling_u` axis maxes at the full trace, so any
+>    config that sustains more than the window's offered rate cannot be measured.
+> 2. **Smoke regime.** This run inherited `replay_time_scale=0.1` +
+>    `max_requests_per_probe=512` (README: convergence test, *not* a benchmark) —
+>    compressed arrivals distort A and the 512 cap imposes a ~8.4 req/s ceiling.
+>
+> The below-ceiling TP1 (2.90) > TP2 (2.21) ordering *may* be real for this model
+> (Qwen3-30B-A3B is an MoE with ~3B active params → little compute per token → TP
+> adds all-reduce overhead with little benefit), which differs from the dense
+> Qwen3.5-27B where TP2 wins. But this run cannot establish it. A valid benchmark
+> needs `scale=1.0`, no cap, and enough offered-load headroom that strong configs
+> are not trace-saturated — see the 27B TP A/B follow-up.
+
+## Phase-5 acceptance
+
+- **No regression.** The primary metric `request_rate_per_gpu` stayed 2.90 the whole
+  run. Scaling TP/DP raised *raw* throughput (4.42, 8.37) but lowered per-GPU
+  efficiency (2.21, 2.09); the loop correctly kept the TP1 baseline as incumbent and
+  never adopted a worse-per-GPU config. (Matches the paper: long-prompt, low-reuse
+  chat prefers small TP for per-GPU efficiency.)
+- **Stop-B authority validated live.** At trial 4 the LLM tried to stop
+  (`should_stop=true`); the deterministic validator **vetoed** it
+  (`validator_did_not_authorize_stop`, `continue_harness_guided_search`), forcing one
+  more confirmation (DP2, which also failed to beat baseline). After the budget, the
+  LLM's later, well-justified stop was honored (`stop_authorized_by:
+  llm_after_veto_budget`). The bounded veto behaved exactly as designed.
+- **Auditable reason chain.** Every stop/veto carries a diagnosis grounded in the
+  measured evidence (e.g. *"increasing TP 1→2 lowers per-GPU efficiency even though
+  token latency improves … EP is explicitly blocked by launch-failure evidence"*).
+- **Launch-failure robustness.** Two EP configs (trial-0003, 0006) failed to launch
+  under vLLM 0.11.1; the harness recorded them as hard-negative evidence and the LLM
+  explicitly stopped proposing EP.
+
+## Notes / limitations
+
+- For this workload the baseline (TP1) is already per-GPU optimal, so iterations-to-
+  *best* = 1; the remaining trials are the loop *confirming* no config beats baseline
+  before stopping. A workload with an under-tuned default would show an improving
+  trajectory; this run validates the stop/no-regression machinery, not a tuning win.
+- The final stop came via `llm_after_veto_budget` (validator vetoed once, then
+  deferred), not a pure deterministic validator stop — because the deterministic
+  conditions (3-within-2%, saturation, validation-exhausted) did not cleanly fire
+  when every trial was a distinct config with a distinct per-GPU rate. The validator
+  acted as the *guard* (preventing premature stop), which is its designed role.
+- 7 trials > the paper's 3–6 average, inflated by the wider search range, 2 EP
+  launch-failures, and the veto. Acceptable for a validation run.
+- LLM token: the non-interactive shell lacks `OPENAI_API_KEY`; export it from the
+  codex `auth.json` (`~/.codex/auth.json`) before the run.

docs/harness-ablation/stop-b-e2e-27b-20260616.md

@@ -0,0 +1,64 @@
+# Stop-B end-to-end on dense Qwen3.5-27B (the improving trajectory) — 2026-06-16
+
+Branch `feat/two-stop`. Real `gpt-5.4` agentic loop (codex/prism), Stop-A enabled,
+length-aware TTFT SLO (4s + L_in/8k, TPOT ≤ 50 ms), vLLM 0.11.1, H20, GPUs 2–7,
+`replay_time_scale=1.0`, `search.high=0.25`, `inherit_incumbent_floor=true`.
+Config `dash0_qwen27b_stopB_loop.json`. Companion to the 30B run
+(`stop-b-e2e-20260616.md`); together they cover all Stop-B behaviors.
+
+## Trajectory (incumbent = TP4 @ 1.00 req/s/GPU)
+
+| iter | proposed by | config | per_gpu | adopted? |
+| --- | --- | --- | --- | --- |
+| 1 | baseline | TP1 | 0.123 | incumbent |
+| 2 | gpt-5.4 | TP2 | 0.2925 (2.4×) | ✅ new incumbent |
+| 3 | gpt-5.4 | TP4 | **1.0012 (8.1×)** | ✅ new incumbent |
+| 4 | gpt-5.4 | TP4 + chunked-prefill + mbt=16384 | 0.942 | ❌ **worse → rejected** |
+| 5 | gpt-5.4 | TP2 + DP2 | (loop stopped before completing) | — |
+
+(Run stopped manually after iter-4 — see "Why stopped" below. Incumbent preserved
+at TP4.)
+
+## What this demonstrates (the piece the 30B run could not)
+
+- **A genuine improving climb.** `gpt-5.4` + the harness raised per-GPU throughput
+  TP1 → TP2 → TP4 (0.123 → 0.29 → 1.00, 8.1×), each step a correctly-diagnosed real
+  gain: TP1 is TPOT-bound, so the agent scaled tensor-parallelism, then — once
+  topology was won — pivoted to **runtime tuning on the winning family** (chunked
+  prefill + larger batched tokens).
+- **No regression.** The runtime tweak (iter-4) measured *below* plain TP4
+  (0.942 < 1.00), and the harness correctly **kept TP4 as the incumbent** rather than
+  adopting the worse config — the core Stop-B guarantee, shown live.
+- Combined with the 30B run (search-high-saturation `validator`-authorized stop +
+  premature-LLM-stop veto), every Stop-B behavior is now validated end-to-end:
+  improving climb, correct bottleneck-driven proposals, no regression, deterministic
+  stop authority, and the LLM-stop veto.
+
+## Process wins / findings
+
+- **SIGTERM teardown fix validated in practice.** This loop was stopped with a plain
+  SIGTERM and the engine + EngineCore workers torn down cleanly — GPUs 2–7 freed, no
+  orphan, no leaked memory (contrast: the pre-fix runs twice leaked GPU0/1). Commit
+  `b17b213`.
+- **Timeout-as-failed-request fix** (`2fcaf80`) held — no trial crashed on
+  request timeouts this run.
+
+## Why stopped (efficiency finding — feeds next round)
+
+The loop was stopped after iter-4 rather than run to an explicit Stop-B firing,
+because each TP4-family trial took ~3 h: at `scale=1.0`, infeasible high-θ probes
+each run to the **`early_stop_max_elapsed_s=900` cap** (`probe_elapsed_s>900`), and
+the primary+fallback binary search doubles the probe count. Stop-A truncates a
+*converged* replay but does not shortcut an *overloaded* probe that simply runs out
+the clock. **For a practical agentic loop at scale=1.0, lower `early_stop_max_elapsed_s`
+(≈300 s)** so overloaded probes die fast; consider also having an infeasible-and-
+overloaded probe early-stop on a fast lag/throughput signal rather than the elapsed
+cap. The convergence itself was already evident (iter-4's runtime tweak and the
+queued TP2+DP2 were not beating TP4).
+
+## Next
+
+- Lower the elapsed cap and (optionally) re-run to capture the explicit Stop-B stop
+  on this 27B stack.
+- Land the deferred items: more robust C feature for the low-reuse regime; Stop-C
+  cross-day retune trigger; §7 baselines (SCOOT/naive/community).

docs/two-stop-summary.md

@@ -0,0 +1,63 @@
+# Two-stop work — summary (feat/two-stop)
+
+Aligns the tuning harness with paper.pdf by implementing and validating the two
+distinct stopping mechanisms the paper conflates, plus a length-aware SLO and two
+harness-robustness fixes. 117 unit tests pass.
+
+## The two stops (they are different things)
+
+- **Stop-A — evaluation sufficiency** (per replay / probe): how much trace to replay
+  before a measurement is trustworthy. Criterion: the replayed prefix's offered
+  L-C-A converges to the full set's. Saves per-evaluation GPU time.
+- **Stop-B — tuning convergence** (across iterations): whether any config-improvement
+  opportunity remains; stop if not. Saves iterations.
+
+## What was built
+
+| Area | Summary | Commit |
+| --- | --- | --- |
+| Unify L-C-A | The prompt's `workload_lca_profile` is now the canonical 10-dim `lca.py` vector, not an ad-hoc re-derivation | `6f8e3c9` |
+| Session-coherent load axis | `sampling_u` assigned per session (parent_chat_id chain) so thresholding keeps/drops whole multi-turn sessions, preserving intra-session KV reuse (C) across load levels | `0f15bbc` |
+| Stop-A | `lca.find_convergence_prefix` (deterministic offered-L-C-A convergence prefix + C-gate: never declare infeasible on a cold cache); `spec.AdaptiveStopSpec` (default off); `worker._adaptive_replay_set` truncates replay + writes a certificate | `51a9e4a` |
+| Stop-A boundary guard | Re-measure the full window when a truncated probe's pass-rate is within `boundary_delta` of the SLO target (fixes feasibility-knee false positives) | `dfc823f` |
+| Stop-B authority | `harness._stop_authority` (mirrors the deterministic validator); `study tune` honors an LLM `should_stop` only if the validator agrees, else a bounded veto | `a8f9034` |
+| Length-aware SLO | `linear_ms` rule: `threshold = intercept_ms + per_token_ms·L_in` (e.g. 4s + L_in/8k) | `ed2bbe0` |
+| Robustness: timeout | HTTP stream now wraps `OSError`/`TimeoutError` as `HttpClientError` — a request exceeding `request_timeout_s` is a failed request, not a crashed trial | `2fcaf80` |
+| Robustness: SIGTERM | `run_trial` installs a SIGTERM handler so killing `study tune` tears down the engine (and EngineCore workers) instead of orphaning it / leaking GPU memory | `b17b213` |
+| Tooling | `stop_a_calibration.py` (CPU convergence curve), `stop_a_validate.py` (offline truncation-fidelity) | `08e53fd`, `3af1d84` |
+
+A subagent code review found no blocking bugs (and independently validated session
+coherence against a real trace); three minor fixes applied in `43125f4`.
+
+## Validation results (real GPU runs, dash0 H20)
+
+- **Stop-A** (`stop-a-validation-20260615.md`): CPU calibration reproduces the paper's
+  C-slowest ordering; GPU fidelity check on Qwen3-30B-A3B saves ~52% replay at τ_c=0.90
+  with 3/4 probe verdicts preserved; the one mismatch is a feasibility-knee false
+  positive that the **boundary guard fixes** — with the guard, best threshold matches
+  full replay exactly while still saving ~38% replay.
+- **27B TP sweep** (`qwen27b-tp-sweep-20260616.md`): under the length-aware SLO, dense
+  Qwen3.5-27B per-GPU rises sharply with TP — TP1 0.065 → TP2 0.29 → TP4 ≥0.91 — the
+  **opposite** of the MoE 30B-A3B (TP1 best per-GPU). TP1 is TPOT-bound.
+- **Stop-B** (`stop-b-e2e-20260615.md`, `stop-b-e2e-27b-20260616.md`): the 30B run
+  shows the deterministic `validator` stop + a premature-LLM-stop **veto**; the 27B run
+  (real gpt-5.4 loop) shows a genuine **improving climb** TP1 0.123 → TP2 0.29 → TP4
+  1.00 req/s/GPU (8.1×), each a correctly-diagnosed gain, then correctly **rejecting** a
+  TP4 runtime tweak that measured worse (no regression). The SIGTERM fix was validated
+  in practice (clean teardown, no leak).
+
+## Open items (next round)
+
+- **Harness convergence ablation (NOT yet done on this branch).** The paper's harness
+  result — domain-knowledge knob-family rules steering the LLM and cutting iterations —
+  has only *qualitative* evidence here (the 27B climb shows correct steering) plus older
+  smoke-regime ablations (`qwen27b-chat-0-8k-harness-fig18.md`: iters-to-best 4→2). A
+  controlled `use_harness=true` vs `false` (naive tuner) comparison on the 27B is the
+  missing quantified result.
+- **Loop efficiency**: at scale=1.0 infeasible high-θ probes burn the
+  `early_stop_max_elapsed_s=900` cap (a TP4 trial took ~3h). Lower it to ~300s for
+  practical agentic loops.
+- **dash0 GPUs 0/1** still hold leaked memory (pre-fix orphans) — needs a root
+  `nvidia-smi --gpu-reset`.
+- Deferred: more robust C feature for the low-reuse regime; Stop-C cross-day retune
+  trigger (paper Q1); §7 baselines (SCOOT / naive / community).

scripts/prepare_trace_windows.py

@@ -92,17 +92,39 @@ def parse_args() -> argparse.Namespace:
     return parser.parse_args()
 
 
-def stable_uniform(*, seed: int, window_id: str, index: int, row: dict[str, Any]) -> float:
+def resolve_session_root(row: dict[str, Any], root_of: dict[Any, Any]) -> Any:
+    """Resolve the session root chat_id for a trace row.
+
+    Sessions are multi-turn chains linked via parent_chat_id (turn>1 points to the
+    parent turn's chat_id, the root turn has parent_chat_id=-1). Because parent
+    turns precede their children in time, a single streaming pass that records
+    chat_id -> root resolves the full chain. Rows whose parent is not yet known
+    (e.g. it fell outside the materialized span) fall back to the parent id so
+    siblings still group together.
+    """
+    chat_id = row.get("chat_id")
+    parent = row.get("parent_chat_id")
+    parent_is_root = (
+        parent is None
+        or (isinstance(parent, (int, float)) and not isinstance(parent, bool) and int(parent) < 0)
+    )
+    root = chat_id if parent_is_root else root_of.get(parent, parent)
+    if chat_id is not None:
+        root_of[chat_id] = root
+    return root
+
+
+def session_uniform(*, seed: int, window_id: str, session_root: Any) -> float:
+    """Deterministic per-session uniform score in [0, 1).
+
+    All turns of a session share one score, so thresholding sampling_u keeps or
+    drops whole sessions and preserves intra-session prefix (KV-cache) reuse.
+    """
     payload = json.dumps(
         {
             "seed": seed,
             "window_id": window_id,
-            "index": index,
-            "timestamp": row.get("timestamp"),
-            "input_length": row.get("input_length"),
-            "output_length": row.get("output_length"),
-            "chat_id": row.get("chat_id"),
-            "turn": row.get("turn"),
+            "session_root": session_root,
         },
         sort_keys=True,
         separators=(",", ":"),
@@ -241,12 +263,16 @@ def materialize_windows(
             bucket = grouped[(trace_path, prompt_path)]
             bucket.sort(key=lambda item: (float(item["window_start"]), str(item["window_id"])))
             matched_rows = 0
+            root_of: dict[Any, Any] = {}
             with trace_path.open() as trace_handle, prompt_path.open() as prompt_handle:
                 for trace_raw, prompt_raw in zip(trace_handle, prompt_handle):
                     trace_raw = trace_raw.strip()
                     if not trace_raw:
                         continue
                     trace_row = json.loads(trace_raw)
+                    # Resolve session linkage for every row (even unmatched ones)
+                    # so multi-turn chains crossing the window edge still group.
+                    session_root = resolve_session_root(trace_row, root_of)
                     timestamp = float(trace_row.get("timestamp") or 0.0)
                     matched_window: dict[str, Any] | None = None
                     for window in bucket:
@@ -267,11 +293,11 @@ def materialize_windows(
                     start = float(matched_window["window_start"])
                     out["source_timestamp"] = timestamp
                     out["timestamp"] = timestamp - start
-                    out["sampling_u"] = stable_uniform(
+                    out["session_root"] = session_root
+                    out["sampling_u"] = session_uniform(
                         seed=sample_seed,
                         window_id=window_id,
-                        index=stats_by_window[window_id].num_requests,
-                        row=merged,
+                        session_root=session_root,
                     )
                     handles[window_id].write(json.dumps(out, ensure_ascii=False) + "\n")
                     stats_by_window[window_id].record(out)
@@ -311,7 +337,7 @@ def build_output_window(
     output["num_excluded_too_long"] = 0
     output["sampling_u_field"] = "sampling_u"
     output["sampling_seed"] = int(sample_seed)
-    output["sampling_strategy"] = "fixed_uniform_score"
+    output["sampling_strategy"] = "session_coherent_uniform_score"
     output["first_request_ts"] = stats.first_request_ts
     output["last_request_ts"] = stats.last_request_ts
     output["first_request_index"] = stats.first_request_index

scripts/run_baseline_then_llm.py

@@ -10,6 +10,7 @@ from aituner.llm import (
     load_capability_profile,
     parse_proposal_text,
 )
+from aituner.lca import build_study_workload_profile
 from aituner.spec import load_study_spec
 from aituner.store import StudyStore
 from aituner.trace import load_trace_requests, summarize_window
@@ -89,6 +90,7 @@ def main() -> int:
             window_summary=summarize_window(requests, window),
             state=state,
             capability_profile=capability_profile,
+            workload_profile=build_study_workload_profile(study, requests, window),
         )
         prompt_name = f"prompt-{state.next_trial_index:04d}"
         store.write_prompt(study.study_id, prompt_name, prompt)

scripts/stop_a_calibration.py

@@ -0,0 +1,128 @@
+#!/usr/bin/env python3
+"""Stop-A calibration: print the offered-L-C-A convergence curve for a raw trace window.
+
+The convergence of prefix-vs-full L-C-A is a deterministic property of the trace
+metadata (lengths, hash_ids, arrivals), so this runs on CPU without serving the
+model. Use it to pick tau / tau_c / stable_checks and to compare how late the C
+dimension converges across workloads (e.g. low-reuse chat vs high-reuse coder).
+
+Example:
+  PYTHONPATH=src python3 scripts/stop_a_calibration.py \
+    --jsonl /dashscope/.../qwen_chat_blksz_64_032109-032111.jsonl \
+    --block-size 64 --window-start 3600 --window-end 4200 --gpu-count 8 --label chat
+"""
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+
+from aituner.lca import find_convergence_prefix, resolve_length_mode
+from aituner.trace import TraceRequest, WindowRecord
+
+
+def _session_root(row: dict, root_of: dict) -> object:
+    chat_id = row.get("chat_id")
+    parent = row.get("parent_chat_id")
+    parent_is_root = parent is None or (
+        isinstance(parent, (int, float)) and not isinstance(parent, bool) and int(parent) < 0
+    )
+    root = chat_id if parent_is_root else root_of.get(parent, parent)
+    if chat_id is not None:
+        root_of[chat_id] = root
+    return root
+
+
+def load_window(jsonl: Path, *, window_start: float, window_end: float) -> list[TraceRequest]:
+    root_of: dict = {}
+    requests: list[TraceRequest] = []
+    with jsonl.open(encoding="utf-8") as handle:
+        for line in handle:
+            line = line.strip()
+            if not line:
+                continue
+            row = json.loads(line)
+            _session_root(row, root_of)  # keep chain complete even outside the window
+            ts = float(row.get("timestamp") or 0.0)
+            if not (window_start <= ts < window_end):
+                continue
+            hash_ids = row.get("hash_ids")
+            requests.append(
+                TraceRequest(
+                    row_id=str(row.get("chat_id")),
+                    arrival_s=ts - window_start,
+                    sampling_u=1.0,
+                    body={},
+                    prompt_tokens_hint=int(row.get("input_length") or 0),
+                    completion_tokens_hint=int(row.get("output_length") or 0),
+                    metadata={"hash_ids": hash_ids if isinstance(hash_ids, list) else None},
+                )
+            )
+    requests.sort(key=lambda item: item.arrival_s)
+    return requests
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--jsonl", type=Path, required=True)
+    ap.add_argument("--block-size", type=int, required=True)
+    ap.add_argument("--window-start", type=float, default=3600.0)
+    ap.add_argument("--window-end", type=float, default=4200.0)
+    ap.add_argument("--gpu-count", type=int, default=8)
+    ap.add_argument("--length-mode", default="total")
+    ap.add_argument("--label", default="")
+    ap.add_argument("--tau", type=float, default=0.9)
+    ap.add_argument("--max-checks", type=int, default=20)
+    args = ap.parse_args()
+
+    requests = load_window(
+        args.jsonl, window_start=args.window_start, window_end=args.window_end
+    )
+    window = WindowRecord(
+        window_id=args.label or args.jsonl.stem,
+        trace_path=args.jsonl,
+        trace_type=args.label or "chat",
+        window_start=0.0,
+        window_end=float(args.window_end - args.window_start),
+        source_payload={"block_size": args.block_size},
+    )
+    mode = resolve_length_mode(length_mode=args.length_mode)
+    rows_with_hash = sum(1 for r in requests if r.metadata.get("hash_ids"))
+    print(
+        f"[{args.label}] requests={len(requests)} rows_with_hash_ids={rows_with_hash} "
+        f"window={args.window_start:.0f}-{args.window_end:.0f}s block_size={args.block_size}"
+    )
+
+    # Full curve (tau_c high so it never short-circuits; we read the curve directly).
+    point = find_convergence_prefix(
+        requests, window, gpu_count=args.gpu_count, length_mode=mode,
+        tau=args.tau, tau_c=1.01, stable_checks=10_000, max_checks=args.max_checks,
+        min_fraction=0.05,
+    )
+    print("  frac   time_s     L      C      A")
+    for c in point.checks:
+        s = c["family_similarity"]
+        print(
+            f"  {c['fraction']:.2f}  {c['time_s']:7.1f}  "
+            f"{s['L']:.3f}  {s['C']:.3f}  {s['A']:.3f}"
+        )
+
+    # Stop fraction at candidate tau_c values (L,A >= tau, C >= tau_c, stable for W=3).
+    print("  -- stop fraction (tau_L=tau_A=%.2f, W=3) --" % args.tau)
+    for tau_c in (0.85, 0.90, 0.92, 0.95):
+        p = find_convergence_prefix(
+            requests, window, gpu_count=args.gpu_count, length_mode=mode,
+            tau=args.tau, tau_c=tau_c, stable_checks=3, max_checks=args.max_checks,
+            min_fraction=0.05,
+        )
+        verdict = (
+            f"stop@frac={p.fraction:.2f} t={p.stop_time_s:.0f}s"
+            if p.converged
+            else "NEVER (replays full window)"
+        )
+        print(f"  tau_c={tau_c:.2f}: {verdict}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/stop_a_validate.py

@@ -0,0 +1,101 @@
+#!/usr/bin/env python3
+"""Validate Stop-A truncation fidelity from a full-replay trial's probe_details.
+
+Given a completed trial that replayed the full window (adaptive_stop disabled), for
+each probe recompute the L-C-A convergence prefix and compare the feasibility
+verdict / pass-rate of the truncated prefix against the full probe. This answers:
+"would Stop-A have changed the measured peak-sustainable-rate?" using only the one
+full run (no second GPU run needed).
+
+Example:
+  PYTHONPATH=src python3 scripts/stop_a_validate.py \
+    --spec configs/examples/dash0_qwen30b_a3b_stopA_fulldata.json \
+    --store-root .aituner/stopA-fulldata --tau 0.9 --tau-c 0.90
+"""
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+
+from aituner.lca import find_convergence_prefix, resolve_length_mode
+from aituner.spec import load_study_spec
+from aituner.trace import load_trace_requests, select_requests_for_threshold
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--spec", type=Path, required=True)
+    ap.add_argument("--store-root", type=Path, required=True)
+    ap.add_argument("--tau", type=float, default=0.9)
+    ap.add_argument("--tau-c", type=float, default=0.90)
+    ap.add_argument("--stable-checks", type=int, default=3)
+    ap.add_argument("--target-pass-rate", type=float, default=0.95)
+    args = ap.parse_args()
+
+    study = load_study_spec(args.spec)
+    window, requests = load_trace_requests(study, study_spec_path=args.spec)
+    mode = resolve_length_mode(request_mode=study.trace.request_mode)
+    gpu_count = study.hardware.gpu_count
+
+    detail_files = sorted(args.store_root.glob("*/trials/*/probe_details.jsonl"))
+    if not detail_files:
+        print(f"no probe_details.jsonl under {args.store_root}")
+        return 1
+
+    print(f"target_pass_rate={args.target_pass_rate} tau={args.tau} tau_c={args.tau_c}")
+    print(
+        "thresh   n_full  stop_idx  frac  full_pass  prefix_pass  "
+        "full_feas  prefix_feas  verdict_match"
+    )
+    mismatches = 0
+    total = 0
+    saved_fractions = []
+    for detail_file in detail_files:
+        with detail_file.open(encoding="utf-8") as handle:
+            for line in handle:
+                line = line.strip()
+                if not line:
+                    continue
+                probe = json.loads(line)
+                threshold = float(probe["threshold"])
+                outcomes = probe.get("outcomes") or []
+                # arrival-ordered outcomes that carry an arrival_s and verdict
+                ordered = sorted(
+                    (o for o in outcomes if o.get("arrival_s") is not None),
+                    key=lambda o: float(o["arrival_s"]),
+                )
+                n = len(ordered)
+                if n == 0:
+                    continue
+                selected = select_requests_for_threshold(requests, threshold=threshold)
+                cp = find_convergence_prefix(
+                    selected, window, gpu_count=gpu_count, length_mode=mode,
+                    tau=args.tau, tau_c=args.tau_c, stable_checks=args.stable_checks,
+                )
+                # Map the convergence prefix fraction onto the replayed outcomes.
+                stop_n = max(1, min(n, round(cp.fraction * n)))
+                full_pass = sum(1 for o in ordered if o.get("evaluation")) / n
+                prefix_pass = sum(1 for o in ordered[:stop_n] if o.get("evaluation")) / stop_n
+                full_feas = full_pass >= args.target_pass_rate
+                prefix_feas = prefix_pass >= args.target_pass_rate
+                match = full_feas == prefix_feas
+                total += 1
+                mismatches += 0 if match else 1
+                saved_fractions.append(1.0 - cp.fraction)
+                print(
+                    f"{threshold:.5f}  {n:6d}  {stop_n:7d}  {cp.fraction:.2f}  "
+                    f"{full_pass:.3f}      {prefix_pass:.3f}        "
+                    f"{str(full_feas):5s}      {str(prefix_feas):5s}        {match}"
+                )
+    if total:
+        avg_saved = sum(saved_fractions) / len(saved_fractions)
+        print(
+            f"\nverdict matches: {total - mismatches}/{total}  "
+            f"mean replay saved: {avg_saved*100:.0f}%"
+        )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/aituner/cli.py

@@ -14,6 +14,7 @@ from .harness import (
 )
 from .job import append_job, build_trial_job
 from .lca import (
+    build_study_workload_profile,
     build_workload_profile,
     resolve_length_mode,
     similarity_report,
@@ -140,6 +141,7 @@ def cmd_study_prompt(args: argparse.Namespace) -> int:
         window_summary=summarize_window(requests, window),
         state=state,
         capability_profile=capability_profile,
+        workload_profile=build_study_workload_profile(study, requests, window),
     )
     prompt_name = args.prompt_name or f"prompt-{state.next_trial_index:04d}"
     path = store.write_prompt(study.study_id, prompt_name, prompt)
@@ -160,6 +162,7 @@ def cmd_study_llm_propose(args: argparse.Namespace) -> int:
         window_summary=summarize_window(requests, window),
         state=state,
         capability_profile=capability_profile,
+        workload_profile=build_study_workload_profile(study, requests, window),
     )
     proposal_text = call_llm_for_proposal(
         policy=study.llm,
@@ -223,6 +226,8 @@ def cmd_study_tune(args: argparse.Namespace) -> int:
     if proposal_files and max_trials > len(proposal_files):
         max_trials = len(proposal_files)
     executed: list[dict[str, object]] = []
+    stop_vetoes = 0
+    max_llm_stop_vetoes = 1
     for idx in range(max_trials):
         state = store.load_state(study.study_id)
         if state.tuning_stop_reason:
@@ -242,11 +247,13 @@ def cmd_study_tune(args: argparse.Namespace) -> int:
             break
         window, requests = load_trace_requests(study, study_spec_path=spec_path)
         window_summary = summarize_window(requests, window)
+        workload_profile = build_study_workload_profile(study, requests, window)
         harness_context = (
             build_harness_context(
                 study=study,
                 window_summary=window_summary,
                 state=state,
+                workload_profile=workload_profile,
             )
             if study.llm.use_harness
             else None
@@ -256,6 +263,7 @@ def cmd_study_tune(args: argparse.Namespace) -> int:
             window_summary=window_summary,
             state=state,
             capability_profile=capability_profile,
+            workload_profile=workload_profile,
         )
         prompt_name = f"prompt-{state.next_trial_index:04d}"
         store.write_prompt(study.study_id, prompt_name, prompt)
@@ -328,7 +336,34 @@ def cmd_study_tune(args: argparse.Namespace) -> int:
         proposal = parse_proposal_text(proposal_text, study)
         store.write_proposal(study.study_id, proposal_name, proposal)
         if proposal.should_stop:
-            if proposal_name.startswith("harness-stop-"):
+            is_harness_stop = proposal_name.startswith("harness-stop-")
+            is_llm_stop = not is_harness_stop and proposal_source is None
+            stop_authority = (
+                harness_context.get("stop_authority")
+                if isinstance(harness_context, dict)
+                else None
+            )
+            authorized = stop_authority is None or bool(stop_authority.get("authorized"))
+            # Stop-B authority: the deterministic validator overrides an
+            # LLM-originated stop. Veto an unauthorized stop (bounded) so the
+            # loop does not converge prematurely on the agent's say-so alone.
+            if is_llm_stop and not authorized and stop_vetoes < max_llm_stop_vetoes:
+                stop_vetoes += 1
+                executed.append(
+                    {
+                        "trial_id": None,
+                        "proposal_name": proposal_name,
+                        "proposal_source": "llm",
+                        "stop_vetoed": True,
+                        "reason": "validator_did_not_authorize_stop",
+                        "validator_reason": (
+                            stop_authority.get("reason") if stop_authority else None
+                        ),
+                        "diagnosis": proposal.diagnosis,
+                    }
+                )
+                continue
+            if is_harness_stop:
                 proposal_source_label = "harness"
             else:
                 proposal_source_label = str(proposal_source) if proposal_source else "llm"
@@ -338,6 +373,13 @@ def cmd_study_tune(args: argparse.Namespace) -> int:
                     "proposal_name": proposal_name,
                     "proposal_source": proposal_source_label,
                     "stopped": True,
+                    "stop_authorized_by": (
+                        "validator"
+                        if (is_harness_stop or authorized)
+                        else "file_proposal"
+                        if proposal_source is not None
+                        else "llm_after_veto_budget"
+                    ),
                     "diagnosis": proposal.diagnosis,
                     "state_best_trial_id": state.best_trial_id,
                     "state_best_request_rate": state.best_request_rate,

src/aituner/harness.py

@@ -4,6 +4,7 @@ import json
 from pathlib import Path
 from typing import Any
 
+from .lca import EPSILON, WorkloadProfile
 from .spec import ConfigPatch, Proposal, StudySpec, StudyState, TrialSummary
 
 
@@ -30,6 +31,7 @@ def build_harness_context(
     study: StudySpec,
     window_summary: dict[str, Any],
     state: StudyState,
+    workload_profile: WorkloadProfile | None = None,
 ) -> dict[str, Any]:
     recent_diagnostics = _recent_trial_diagnostics(state)
     trial_profiles = _trial_profiles(study, recent_diagnostics)
@@ -46,24 +48,32 @@ def build_harness_context(
         trial_profiles,
         bottleneck_hypotheses,
     )
+    harness_stop = _harness_stop_decision(
+        study,
+        state,
+        recent_diagnostics,
+        experiment_plan=experiment_plan,
+    )
     return {
         "paper_alignment": {
             "goal": "Use workload-feature-to-knob harnesses to reduce wasted trials and avoid regressing after a good configuration is found.",
             "feature_model": "L-C-A: request lengths, inter-request KV-cache reuse, and arrival dynamics.",
             "trial_policy": "Profile measured trials, rank bottleneck hypotheses, score generic candidate actions, and stop only when no useful measured hypothesis remains.",
         },
-        "workload_lca_profile": _workload_lca_profile(window_summary),
+        "workload_lca_profile": _workload_lca_profile(window_summary, workload_profile),
         "recent_trial_diagnostics": recent_diagnostics,
         "trial_profiles": trial_profiles,
         "bottleneck_hypotheses": bottleneck_hypotheses,
         "candidate_actions": experiment_plan["candidate_actions"],
         "experiment_plan": experiment_plan,
         "convergence_guard": _convergence_guard(state, recent_diagnostics),
-        "harness_stop": _harness_stop_decision(
+        "harness_stop": harness_stop,
+        "stop_authority": _stop_authority(
             study,
             state,
             recent_diagnostics,
-            experiment_plan=experiment_plan,
+            experiment_plan,
+            harness_stop,
         ),
         "harness_proposal": _harness_proposal_decision(
             study,
@@ -141,7 +151,12 @@ def render_harness_context(context: dict[str, Any]) -> str:
     return json.dumps(context, ensure_ascii=False, indent=2)
 
 
-def _workload_lca_profile(window_summary: dict[str, Any]) -> dict[str, Any]:
+def _workload_lca_profile(
+    window_summary: dict[str, Any],
+    workload_profile: WorkloadProfile | None = None,
+) -> dict[str, Any]:
+    if workload_profile is not None:
+        return _canonical_lca_profile(workload_profile)
     prefix_cache = window_summary.get("prefix_cache")
     if not isinstance(prefix_cache, dict):
         prefix_cache = {}
@@ -178,6 +193,54 @@ def _workload_lca_profile(window_summary: dict[str, Any]) -> dict[str, Any]:
     }
 
 
+def _canonical_lca_profile(profile: WorkloadProfile) -> dict[str, Any]:
+    """Authoritative L-C-A block: the paper's 10-dim RobustScaler vector.
+
+    Sourced from lca.WorkloadProfile so the prompt's L-C-A is the same metric
+    used for the workload-similarity computations, not an ad-hoc re-derivation.
+    The regime labels reuse the heuristic classifiers but are fed from the
+    canonical stats.
+    """
+    stats = profile.stats if isinstance(profile.stats, dict) else {}
+    length = stats.get("length") if isinstance(stats.get("length"), dict) else {}
+    cache = stats.get("cache") if isinstance(stats.get("cache"), dict) else {}
+    arrival = stats.get("arrival") if isinstance(stats.get("arrival"), dict) else {}
+    length_p95 = _as_float(length.get("p95"))
+    length_p50 = _as_float(length.get("p50"))
+    tail_ratio = float(length_p95 / max(length_p50, EPSILON)) if length_p95 else 0.0
+    repeated_token_ratio = _as_float(cache.get("input_hit_rate"))
+    fano_1s = _as_float(arrival.get("fano_1s"))
+    interarrival_cv = _as_float(arrival.get("interarrival_cv"))
+    return {
+        "metric": "paper L-C-A (10-dim, RobustScaler-normalized) from lca.WorkloadProfile",
+        "length_mode": profile.length_mode,
+        "feature_names": list(profile.feature_names),
+        "vector": list(profile.vector),
+        "L_request_lengths": {
+            "mean": _as_float(length.get("mean")),
+            "p50": length_p50,
+            "p95": length_p95,
+            "cv": _as_float(length.get("cv")),
+            "tail_ratio_p95_p50": tail_ratio,
+            "regime": _length_regime(length_p95, tail_ratio),
+        },
+        "C_prefix_cache": {
+            "hit_rate": _as_float(cache.get("hit_rate")),
+            "input_hit_rate": repeated_token_ratio,
+            "repeated_block_ratio": _as_float(cache.get("repeated_block_ratio")),
+            "rows_with_hash_ids": int(cache.get("rows_with_hash_ids") or 0),
+            "regime": _cache_regime(repeated_token_ratio),
+        },
+        "A_arrivals": {
+            "request_rate": _as_float(arrival.get("request_rate")),
+            "request_rate_per_gpu": _as_float(arrival.get("request_rate_per_gpu")),
+            "interarrival_cv": interarrival_cv,
+            "fano_1s": fano_1s,
+            "regime": _arrival_regime(fano_1s, interarrival_cv),
+        },
+    }
+
+
 def _knob_harnesses(
     study: StudySpec,
     window_summary: dict[str, Any],
@@ -753,6 +816,43 @@ def _harness_stop_decision(
     }
 
 
+def _stop_authority(
+    study: StudySpec,
+    state: StudyState,
+    recent_diagnostics: list[dict[str, Any]],
+    experiment_plan: dict[str, Any] | None,
+    harness_stop: dict[str, Any],
+) -> dict[str, Any]:
+    """Stop-B authority: the deterministic validator decides if stopping is justified.
+
+    ``authorized`` mirrors the deterministic harness stop decision. The LLM's
+    should_stop is only a corroborating signal: the tuning loop honors an
+    LLM-originated stop only when this validator authorizes it (or when the
+    harness is disabled). ``opportunity_remains`` flags that a concrete adjacent
+    probe (open topology frontier or a high-value planned candidate) still exists,
+    so an early stop would leave measured headroom on the table.
+    """
+    frontier = _topology_frontier_status(study, state, recent_diagnostics)
+    next_action = (
+        experiment_plan.get("next_action") if isinstance(experiment_plan, dict) else None
+    )
+    has_candidate = (
+        isinstance(next_action, dict) and _as_float(next_action.get("score")) >= 0.35
+    )
+    opportunity_remains = bool(frontier.get("frontier_open")) or has_candidate
+    authorized = bool(harness_stop.get("should_stop"))
+    return {
+        "authorized": authorized,
+        "reason": harness_stop.get("reason"),
+        "opportunity_remains": opportunity_remains,
+        "summary": (
+            "Deterministic validator authorizes stop; no adjacent bottleneck probe remains."
+            if authorized
+            else "Validator does not authorize stop; LLM should_stop is advisory only."
+        ),
+    }
+
+
 def _harness_proposal_decision(
     study: StudySpec,
     window_summary: dict[str, Any],

src/aituner/http_client.py

@@ -179,6 +179,9 @@ def chat_completion(
     except urllib.error.HTTPError as exc:
         detail = exc.read().decode("utf-8", errors="replace")
         raise HttpClientError(f"llm_completion failed: {exc.code} {detail}") from exc
+    except OSError as exc:
+        # TimeoutError (socket.timeout), URLError, ConnectionError all subclass OSError.
+        raise HttpClientError(f"llm_completion failed: {exc}") from exc
 
 
 def stream_text_completion(
@@ -232,6 +235,8 @@ def stream_text_completion(
     except urllib.error.HTTPError as exc:
         detail = exc.read().decode("utf-8", errors="replace")
         raise HttpClientError(f"stream_text_completion failed: {exc.code} {detail}") from exc
+    except OSError as exc:
+        raise HttpClientError(f"stream_text_completion failed: {exc}") from exc
     return "".join(parts)
 
 
@@ -293,6 +298,10 @@ def stream_chat_completion(
     except urllib.error.HTTPError as exc:
         detail = exc.read().decode("utf-8", errors="replace")
         raise HttpClientError(f"stream_chat_completion failed: {exc.code} {detail}") from exc
+    except OSError as exc:
+        # A request that exceeds request_timeout_s raises TimeoutError mid-stream;
+        # treat it as a failed request (SLO miss), not a crashed trial.
+        raise HttpClientError(f"stream_chat_completion failed: {exc}") from exc
     ttft_ms = None if first_token_at is None else (first_token_at - start) * 1000.0
     if completion_tokens is None and chunk_token_count > 0:
         completion_tokens = chunk_token_count

src/aituner/lca.py

@@ -4,10 +4,13 @@ import json
 import math
 import statistics
 from dataclasses import dataclass
-from typing import Any, Sequence
+from typing import TYPE_CHECKING, Any, Sequence
 
 from .trace import TraceRequest, WindowRecord
 
+if TYPE_CHECKING:
+    from .spec import StudySpec
+
 
 EPSILON = 1e-9
 
@@ -178,6 +181,28 @@ def build_workload_profile(
     )
 
 
+def build_study_workload_profile(
+    study: "StudySpec",
+    requests: list[TraceRequest],
+    window: WindowRecord,
+) -> WorkloadProfile:
+    """Canonical L-C-A profile for a study's loaded window.
+
+    This is the single source of truth for the paper's 10-dimensional L-C-A
+    feature vector used by the harness prompt and (later) by Stop-A.
+    """
+    mode = resolve_length_mode(
+        request_mode=study.trace.request_mode,
+        length_mode="auto",
+    )
+    return build_workload_profile(
+        requests,
+        window,
+        gpu_count=study.hardware.gpu_count,
+        length_mode=mode,
+    )
+
+
 def fit_robust_scale(profiles: Sequence[WorkloadProfile]) -> RobustScale:
     if not profiles:
         raise ValueError("At least one profile is required to fit a robust scale.")
@@ -234,6 +259,152 @@ def similarity_report(profiles: Sequence[WorkloadProfile]) -> dict[str, Any]:
     }
 
 
+@dataclass(frozen=True)
+class ConvergencePoint:
+    converged: bool
+    stop_index: int
+    stop_time_s: float
+    fraction: float
+    family_similarity: dict[str, float]
+    checks: list[dict[str, Any]]
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "converged": self.converged,
+            "stop_index": self.stop_index,
+            "stop_time_s": self.stop_time_s,
+            "fraction": self.fraction,
+            "family_similarity": self.family_similarity,
+            "checks": self.checks,
+        }
+
+
+def find_convergence_prefix(
+    requests: list[TraceRequest],
+    window: WindowRecord,
+    *,
+    gpu_count: int,
+    length_mode: str = "total",
+    tau: float = 0.9,
+    tau_c: float = 0.92,
+    stable_checks: int = 3,
+    max_checks: int = 20,
+    min_fraction: float = 0.1,
+) -> ConvergencePoint:
+    """Earliest arrival-ordered prefix whose offered L-C-A converges to the full set.
+
+    The L-C-A vector is a deterministic function of the trace metadata, so the
+    convergence of prefix-vs-full is itself deterministic (the paper's Fig. 9
+    curve). Stop-A replays only up to this prefix. A prefix counts as converged
+    when the L and A family similarities reach ``tau`` and the (slowest) C family
+    similarity reaches the stricter ``tau_c`` for ``stable_checks`` consecutive
+    checkpoints. If that never happens within the window the point reports the
+    full set (converged=False), which keeps the C-gate honest: an unconverged C
+    means the probe must replay the whole window rather than stop early.
+    """
+    total = len(requests)
+    if total == 0:
+        return ConvergencePoint(
+            converged=False,
+            stop_index=0,
+            stop_time_s=0.0,
+            fraction=1.0,
+            family_similarity={"L": 1.0, "C": 1.0, "A": 1.0},
+            checks=[],
+        )
+    # Compare each arrival-ordered prefix to the whole set, both measured over
+    # their own elapsed span so the A (rate) dimension is comparable rather than
+    # diluted by the fixed window length.
+    target = _prefix_profile(
+        requests, total, window, gpu_count=gpu_count, length_mode=length_mode
+    )
+    indices = _checkpoint_indices(
+        total, max_checks=max_checks, min_fraction=min_fraction
+    )
+    checks: list[dict[str, Any]] = []
+    consecutive = 0
+    converged_index: int | None = None
+    converged_sims: dict[str, float] | None = None
+    for index in indices:
+        prefix = _prefix_profile(
+            requests, index, window, gpu_count=gpu_count, length_mode=length_mode
+        )
+        sims = _family_similarity(target.vector, prefix.vector)
+        checks.append(
+            {
+                "index": index,
+                "fraction": float(index / total),
+                "time_s": float(requests[index - 1].arrival_s),
+                "family_similarity": sims,
+            }
+        )
+        passed = sims["L"] >= tau and sims["A"] >= tau and sims["C"] >= tau_c
+        consecutive = consecutive + 1 if passed else 0
+        if consecutive >= stable_checks and converged_index is None:
+            converged_index = index
+            converged_sims = sims
+            break
+    if converged_index is None:
+        last_sims = checks[-1]["family_similarity"] if checks else {"L": 1.0, "C": 1.0, "A": 1.0}
+        return ConvergencePoint(
+            converged=False,
+            stop_index=total,
+            stop_time_s=float(requests[-1].arrival_s),
+            fraction=1.0,
+            family_similarity=last_sims,
+            checks=checks,
+        )
+    return ConvergencePoint(
+        converged=True,
+        stop_index=converged_index,
+        stop_time_s=float(requests[converged_index - 1].arrival_s),
+        fraction=float(converged_index / total),
+        family_similarity=converged_sims or {},
+        checks=checks,
+    )
+
+
+def _prefix_profile(
+    requests: list[TraceRequest],
+    index: int,
+    window: WindowRecord,
+    *,
+    gpu_count: int,
+    length_mode: str,
+) -> WorkloadProfile:
+    prefix = requests[:index]
+    start = float(prefix[0].arrival_s) if prefix else float(window.window_start)
+    end = float(prefix[-1].arrival_s) if prefix else float(window.window_start)
+    prefix_window = WindowRecord(
+        window_id=window.window_id,
+        trace_path=window.trace_path,
+        trace_type=window.trace_type,
+        window_start=start,
+        window_end=end,
+        source_payload=window.source_payload,
+    )
+    return build_workload_profile(
+        prefix, prefix_window, gpu_count=gpu_count, length_mode=length_mode
+    )
+
+
+def _checkpoint_indices(total: int, *, max_checks: int, min_fraction: float) -> list[int]:
+    start = max(1, int(math.ceil(min_fraction * total)))
+    if total <= max_checks:
+        candidates = range(start, total + 1)
+    else:
+        step = max(1, total // max_checks)
+        candidates = list(range(start, total + 1, step))
+        if candidates and candidates[-1] != total:
+            candidates.append(total)
+    seen: list[int] = []
+    for value in candidates:
+        clamped = min(total, max(1, int(value)))
+        if not seen or seen[-1] != clamped:
+            seen.append(clamped)
+    return seen
+
+
 def dumps_profile(profile: WorkloadProfile) -> str:
     return json.dumps(profile.to_dict(), ensure_ascii=False, indent=2) + "\n"
 

src/aituner/llm.py

@@ -3,12 +3,15 @@ from __future__ import annotations
 import json
 import time
 from pathlib import Path
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from .harness import build_harness_context, render_harness_context
 from .http_client import chat_completion, stream_text_completion
 from .spec import LLMPolicySpec, Proposal, SpecError, StudySpec, StudyState
 
+if TYPE_CHECKING:
+    from .lca import WorkloadProfile
+
 
 def _parse_bool_like(value: Any, *, context: str) -> bool:
     if isinstance(value, bool):
@@ -178,6 +181,7 @@ def build_prompt(
     window_summary: dict[str, Any],
     state: StudyState,
     capability_profile: dict[str, Any] | None,
+    workload_profile: "WorkloadProfile | None" = None,
 ) -> str:
     objective_notes: list[str] = []
     if study.trace.request_mode == "decode_only":
@@ -409,6 +413,7 @@ def build_prompt(
                     study=study,
                     window_summary=window_summary,
                     state=state,
+                    workload_profile=workload_profile,
                 )
             ),
             "",

src/aituner/slo.py

@@ -29,6 +29,9 @@ def _rule_threshold_ms(rule: ThresholdRule, prompt_tokens: int | None) -> float:
     if rule.kind == "fixed_ms":
         assert rule.threshold_ms is not None
         return rule.threshold_ms
+    if rule.kind == "linear_ms":
+        assert rule.intercept_ms is not None and rule.per_token_ms is not None
+        return float(rule.intercept_ms) + float(rule.per_token_ms) * float(prompt_tokens or 0)
     if rule.kind != "step_ms":
         raise ValueError(f"Unsupported threshold rule: {rule.kind}")
     prompt = float(prompt_tokens or 0)

src/aituner/spec.py

@@ -321,6 +321,71 @@ class InputLengthFilterSpec:
         return spec
 
 
+@dataclass(frozen=True)
+class AdaptiveStopSpec:
+    """Stop-A: truncate per-probe replay once the offered L-C-A converges.
+
+    Disabled by default; the thresholds are calibrated per workload (Phase 3)
+    before being switched on, so existing studies are unaffected.
+    """
+
+    enabled: bool = False
+    tau: float = 0.9
+    tau_c: float = 0.92
+    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":
+        if data is None:
+            return cls()
+        m = _require_mapping(data, context="trace.adaptive_stop")
+        enabled = (
+            _require_bool(m.get("enabled"), context="trace.adaptive_stop.enabled")
+            if m.get("enabled") is not None
+            else False
+        )
+        tau = _require_float(m.get("tau", 0.9), context="trace.adaptive_stop.tau")
+        tau_c = _require_float(m.get("tau_c", 0.92), context="trace.adaptive_stop.tau_c")
+        stable_checks = _require_int(
+            m.get("stable_checks", 3), context="trace.adaptive_stop.stable_checks"
+        )
+        max_checks = _require_int(
+            m.get("max_checks", 20), context="trace.adaptive_stop.max_checks"
+        )
+        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."
+            )
+        if stable_checks > max_checks:
+            raise SpecError(
+                "trace.adaptive_stop.stable_checks must be <= max_checks, "
+                "otherwise convergence can never be detected."
+            )
+        return cls(
+            enabled=enabled,
+            tau=tau,
+            tau_c=tau_c,
+            stable_checks=stable_checks,
+            max_checks=max_checks,
+            min_fraction=min_fraction,
+            boundary_delta=boundary_delta,
+        )
+
+
 @dataclass(frozen=True)
 class TraceSpec:
     windows_path: str
@@ -338,6 +403,7 @@ class TraceSpec:
     early_stop_max_lag_s: float | None = None
     early_stop_max_elapsed_s: float | None = None
     restart_engine_after_early_stop: bool = False
+    adaptive_stop: AdaptiveStopSpec = AdaptiveStopSpec()
 
     @classmethod
     def from_dict(cls, data: Mapping[str, Any]) -> "TraceSpec":
@@ -429,6 +495,7 @@ class TraceSpec:
                 if data.get("restart_engine_after_early_stop") is not None
                 else request_mode == "decode_only"
             ),
+            adaptive_stop=AdaptiveStopSpec.from_dict(data.get("adaptive_stop")),
         )
 
 
@@ -437,6 +504,8 @@ class ThresholdRule:
     kind: str
     threshold_ms: float | None = None
     buckets: list[dict[str, float]] = field(default_factory=list)
+    intercept_ms: float | None = None
+    per_token_ms: float | None = None
 
     @classmethod
     def from_dict(cls, data: Mapping[str, Any], *, context: str) -> "ThresholdRule":
@@ -448,6 +517,18 @@ class ThresholdRule:
                     data.get("threshold_ms"), context=f"{context}.threshold_ms"
                 ),
             )
+        if kind == "linear_ms":
+            # threshold = intercept_ms + per_token_ms * input_tokens
+            # e.g. "4s + L_in/8k" -> intercept_ms=4000, per_token_ms=0.125
+            intercept_ms = _require_float(
+                data.get("intercept_ms"), context=f"{context}.intercept_ms"
+            )
+            per_token_ms = _require_float(
+                data.get("per_token_ms"), context=f"{context}.per_token_ms"
+            )
+            if intercept_ms < 0 or per_token_ms < 0:
+                raise SpecError(f"{context}.intercept_ms/per_token_ms must be >= 0.")
+            return cls(kind=kind, intercept_ms=intercept_ms, per_token_ms=per_token_ms)
         if kind == "step_ms":
             raw = data.get("buckets")
             if not isinstance(raw, list) or not raw:

src/aituner/worker.py

@@ -16,6 +16,7 @@ from typing import Any, Callable
 
 from .engine import build_launch_recipe
 from .http_client import HttpClientError, stream_chat_completion, wait_for_server
+from .lca import find_convergence_prefix, resolve_length_mode
 from .search import ThresholdProbe, binary_search_max_feasible
 from .slo import RequestOutcome, evaluate_request, summarize_evaluations
 from .spec import ConfigPatch, SamplingSearchSpec, TrialSpec, load_study_spec, to_jsonable
@@ -209,6 +210,112 @@ def _probe_outcome_details(
     }
 
 
+_SIGTERM_NOT_INSTALLED = object()
+
+
+def _install_sigterm_as_keyboardinterrupt() -> Any:
+    """Make SIGTERM raise KeyboardInterrupt so the engine-teardown finally runs.
+
+    When `study tune` is killed, a default SIGTERM skips the finally blocks and
+    orphans the vLLM engine (and its EngineCore workers) on the GPUs. Converting
+    SIGTERM to KeyboardInterrupt lets _terminate_process_tree run. Only installable
+    from the main thread; returns the previous handler (or a sentinel).
+    """
+    if threading.current_thread() is not threading.main_thread():
+        return _SIGTERM_NOT_INSTALLED
+
+    def _handler(signum: int, frame: Any) -> None:
+        raise KeyboardInterrupt()
+
+    try:
+        return signal.signal(signal.SIGTERM, _handler)
+    except (ValueError, OSError):
+        return _SIGTERM_NOT_INSTALLED
+
+
+def _restore_sigterm(previous: Any) -> None:
+    if previous is _SIGTERM_NOT_INSTALLED:
+        return
+    if threading.current_thread() is not threading.main_thread():
+        return
+    try:
+        signal.signal(signal.SIGTERM, previous)
+    except (ValueError, OSError):
+        pass
+
+
+def _ignore_sigterm_if_main() -> None:
+    """Ignore SIGTERM during teardown so a second signal cannot orphan the engine."""
+    if threading.current_thread() is not threading.main_thread():
+        return
+    try:
+        signal.signal(signal.SIGTERM, signal.SIG_IGN)
+    except (ValueError, OSError):
+        pass
+
+
+def _adaptive_replay_set(
+    selected: list[TraceRequest],
+    *,
+    study: Any,
+    window: Any,
+) -> tuple[list[TraceRequest], dict[str, Any] | None]:
+    """Stop-A: truncate the replay to the offered-L-C-A convergence prefix.
+
+    Returns the (possibly shortened) request list to replay and a certificate of
+    the convergence decision. When Stop-A is disabled, or C never converges, the
+    full selected set is replayed (the C-gate: no early stop on a cold cache).
+    """
+    spec = study.trace.adaptive_stop
+    if not getattr(spec, "enabled", False) or not selected:
+        return selected, None
+    point = find_convergence_prefix(
+        selected,
+        window,
+        gpu_count=study.hardware.gpu_count,
+        length_mode=resolve_length_mode(request_mode=study.trace.request_mode),
+        tau=spec.tau,
+        tau_c=spec.tau_c,
+        stable_checks=spec.stable_checks,
+        max_checks=spec.max_checks,
+        min_fraction=spec.min_fraction,
+    )
+    replay = selected[: point.stop_index] if point.stop_index > 0 else selected
+    certificate = {
+        "enabled": True,
+        "converged": point.converged,
+        "stop_index": point.stop_index,
+        "total_selected": len(selected),
+        "fraction": point.fraction,
+        "stop_time_s": point.stop_time_s,
+        "family_similarity": point.family_similarity,
+    }
+    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
@@ -505,6 +612,7 @@ def run_trial(trial_spec_path: Path) -> dict[str, Any]:
             )
 
         process = launch_process()
+        previous_sigterm = _install_sigterm_as_keyboardinterrupt()
         probe_history: list[dict[str, Any]] = []
         failure_stage = "engine_launch"
         try:
@@ -519,27 +627,49 @@ def run_trial(trial_spec_path: Path) -> dict[str, Any]:
             def evaluator(threshold: float) -> ThresholdProbe[ProbePayload]:
                 nonlocal process
                 selected = select_requests_for_threshold(requests, threshold=threshold)
-                restart_after_early_stop = study.trace.restart_engine_after_early_stop
-                outcomes, early_stopped, early_stop_reason = _replay_requests(
-                    selected,
-                    base_url=recipe.base_url,
-                    timeout_s=recipe.request_timeout_s,
-                    max_concurrency=study.trace.max_concurrency,
-                    target_pass_rate=study.slo.target_pass_rate,
-                    max_lag_s=study.trace.early_stop_max_lag_s,
-                    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,
+                replay_set, adaptive_stop_certificate = _adaptive_replay_set(
+                    selected, study=study, window=window
                 )
+                restart_after_early_stop = study.trace.restart_engine_after_early_stop
+
+                def _run(reqs: list[TraceRequest]) -> tuple[list[RequestOutcome], bool, str]:
+                    return _replay_requests(
+                        reqs,
+                        base_url=recipe.base_url,
+                        timeout_s=recipe.request_timeout_s,
+                        max_concurrency=study.trace.max_concurrency,
+                        target_pass_rate=study.slo.target_pass_rate,
+                        max_lag_s=study.trace.early_stop_max_lag_s,
+                        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=selected,
+                    selected=replay_set,
                     outcomes=outcomes,
                     evaluations=evaluations,
                     early_stopped=early_stopped,
                     early_stop_reason=early_stop_reason,
                 )
+                probe_details["adaptive_stop"] = adaptive_stop_certificate
                 with probe_details_path.open("a", encoding="utf-8") as details_handle:
                     details_handle.write(
                         json.dumps(probe_details, ensure_ascii=False) + "\n"
@@ -580,12 +710,14 @@ def run_trial(trial_spec_path: Path) -> dict[str, Any]:
                 probe_record = {
                     "threshold": threshold,
                     "request_count": payload.request_count,
+                    "replayed_request_count": len(replay_set),
                     "pass_rate": payload.pass_rate,
                     "request_rate": payload.request_rate,
                     "feasible": payload.feasible,
                     "early_stopped": payload.early_stopped,
                     "early_stop_reason": payload.early_stop_reason,
                     "latency_summary": payload.latency_summary,
+                    "adaptive_stop": adaptive_stop_certificate,
                 }
                 probe_history.append(probe_record)
                 StudyStore.write_json(Path(trial.probe_log_path), probe_history)
@@ -739,4 +871,6 @@ def run_trial(trial_spec_path: Path) -> dict[str, Any]:
             StudyStore.write_json(Path(trial.result_path), result)
             return result
         finally:
+            _ignore_sigterm_if_main()
             _terminate_process_tree(process, timeout_s=30.0, marker_env=trial_marker_env)
+            _restore_sigterm(previous_sigterm)

tests/test_core_flow.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+import contextlib
 import io
 import math
 import os
@@ -15,6 +16,7 @@ from aituner.cli import main as cli_main
 from aituner.compare import _aggregate_summary, load_compare_spec, run_compare
 from aituner.engine import build_launch_recipe
 from aituner.http_client import (
+    HttpClientError,
     StreamMetrics,
     _auth_headers,
     _openai_url,
@@ -28,7 +30,9 @@ from aituner.harness import (
     build_harness_stop_proposal,
 )
 from aituner.lca import (
+    build_study_workload_profile,
     build_workload_profile,
+    find_convergence_prefix,
     profile_similarity,
     resolve_length_mode,
     similarity_report,
@@ -37,9 +41,11 @@ from aituner.llm import _extract_response_text, build_prompt, parse_proposal_tex
 from aituner.search import ThresholdProbe, binary_search_max_feasible
 from aituner.slo import RequestOutcome, evaluate_request, summarize_evaluations
 from aituner.spec import (
+    AdaptiveStopSpec,
     ConfigPatch,
     LLMEndpointSpec,
     Proposal,
+    SloSpec,
     SpecError,
     StudyState,
     TrialSummary,
@@ -48,6 +54,10 @@ from aituner.spec import (
 from aituner.store import StudyStore
 from aituner.trace import load_trace_requests, summarize_window
 from aituner.worker import (
+    _adaptive_replay_set,
+    _install_sigterm_as_keyboardinterrupt,
+    _restore_sigterm,
+    _should_extend_on_boundary,
     _best_feasible_probe_record,
     _latency_summary,
     _run_one_request,
@@ -298,6 +308,301 @@ class CoreFlowTests(unittest.TestCase):
         self.assertAlmostEqual(profile.stats["arrival"]["fano_1s"], 0.5)
         self.assertEqual(resolve_length_mode(request_mode="decode_only"), "output")
 
+    def test_harness_context_uses_canonical_lca_vector(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            tmp_path = Path(tmp)
+            study_path = _write_study_assets(tmp_path)
+            study = load_study_spec(study_path)
+            window, requests = load_trace_requests(study, study_spec_path=study_path)
+            profile = build_study_workload_profile(study, requests, window)
+            state = StudyState(study_id=study.study_id, trials=[])
+            summary = summarize_window(requests, window)
+            context = build_harness_context(
+                study=study,
+                window_summary=summary,
+                state=state,
+                workload_profile=profile,
+            )
+            block = context["workload_lca_profile"]
+            # The labeled L-C-A block is the canonical 10-dim metric, not ad-hoc.
+            self.assertEqual(block["vector"], profile.vector)
+            self.assertEqual(len(block["vector"]), 10)
+            self.assertIn("RobustScaler", block["metric"])
+            # Without a profile it falls back to the legacy ad-hoc rendering.
+            legacy = build_harness_context(
+                study=study,
+                window_summary=summary,
+                state=state,
+            )["workload_lca_profile"]
+            self.assertNotIn("vector", legacy)
+
+    def _steady_requests(self, count: int, *, input_tokens: int = 100) -> list:
+        return [
+            TraceRequest(
+                row_id=f"r{i}",
+                arrival_s=float(i),
+                sampling_u=1.0,
+                body={},
+                prompt_tokens_hint=input_tokens,
+                completion_tokens_hint=16,
+                metadata={"hash_ids": None},
+            )
+            for i in range(count)
+        ]
+
+    def _conv_window(self) -> WindowRecord:
+        return WindowRecord(
+            window_id="conv",
+            trace_path=Path("trace.jsonl"),
+            trace_type="chat",
+            window_start=0.0,
+            window_end=0.0,
+            source_payload={"block_size": 64},
+        )
+
+    def test_convergence_prefix_stops_early_on_stationary_trace(self) -> None:
+        requests = self._steady_requests(60)
+        point = find_convergence_prefix(
+            requests,
+            self._conv_window(),
+            gpu_count=1,
+            length_mode="total",
+            tau=0.9,
+            tau_c=0.9,
+            stable_checks=3,
+            max_checks=20,
+            min_fraction=0.1,
+        )
+        self.assertTrue(point.converged)
+        # A stationary workload should be trustworthy well before the full window.
+        self.assertLess(point.stop_index, len(requests))
+        self.assertLess(point.fraction, 1.0)
+        self.assertTrue(point.checks)
+
+    def test_convergence_prefix_waits_when_cache_warms_late(self) -> None:
+        window = self._conv_window()
+        # First half: no prefix reuse. Second half: every request reuses block 1,
+        # so the C dimension only stabilizes once the reuse regime is exercised.
+        requests = []
+        for i in range(30):
+            requests.append(
+                TraceRequest(
+                    row_id=f"cold{i}",
+                    arrival_s=float(i),
+                    sampling_u=1.0,
+                    body={},
+                    prompt_tokens_hint=640,
+                    completion_tokens_hint=16,
+                    metadata={"hash_ids": [10_000 + i]},
+                )
+            )
+        for i in range(30):
+            requests.append(
+                TraceRequest(
+                    row_id=f"warm{i}",
+                    arrival_s=float(30 + i),
+                    sampling_u=1.0,
+                    body={},
+                    prompt_tokens_hint=640,
+                    completion_tokens_hint=16,
+                    metadata={"hash_ids": [1, 2, 3, 4, 5]},
+                )
+            )
+        point = find_convergence_prefix(
+            requests,
+            window,
+            gpu_count=1,
+            length_mode="total",
+            tau=0.9,
+            tau_c=0.95,
+            stable_checks=2,
+            max_checks=20,
+            min_fraction=0.1,
+        )
+        # The C family similarity must be low while only the cold half is seen.
+        early = [c for c in point.checks if c["fraction"] <= 0.4]
+        self.assertTrue(early)
+        self.assertTrue(any(c["family_similarity"]["C"] < 0.9 for c in early))
+
+    def test_stop_authority_mirrors_validator_and_blocks_fresh_stop(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            study = load_study_spec(_write_study_assets(Path(tmp)))
+            state = StudyState(study_id=study.study_id, trials=[])
+            context = build_harness_context(
+                study=study,
+                window_summary={},
+                state=state,
+            )
+            authority = context["stop_authority"]
+            # The authority is the deterministic validator; with no completed
+            # trials it must not authorize a stop.
+            self.assertEqual(
+                authority["authorized"], context["harness_stop"]["should_stop"]
+            )
+            self.assertFalse(authority["authorized"])
+
+    def test_adaptive_replay_set_truncates_only_when_enabled(self) -> None:
+        from types import SimpleNamespace
+
+        requests = self._steady_requests(60)
+        window = self._conv_window()
+        enabled_study = SimpleNamespace(
+            trace=SimpleNamespace(
+                adaptive_stop=AdaptiveStopSpec(
+                    enabled=True,
+                    tau=0.9,
+                    tau_c=0.9,
+                    stable_checks=3,
+                    max_checks=20,
+                    min_fraction=0.1,
+                ),
+                request_mode="chat",
+            ),
+            hardware=SimpleNamespace(gpu_count=1),
+        )
+        replay, certificate = _adaptive_replay_set(
+            requests, study=enabled_study, window=window
+        )
+        self.assertIsNotNone(certificate)
+        self.assertTrue(certificate["enabled"])
+        self.assertEqual(len(replay), certificate["stop_index"])
+        self.assertLessEqual(len(replay), len(requests))
+
+        disabled_study = SimpleNamespace(
+            trace=SimpleNamespace(
+                adaptive_stop=AdaptiveStopSpec(enabled=False),
+                request_mode="chat",
+            ),
+            hardware=SimpleNamespace(gpu_count=1),
+        )
+        passthrough, no_cert = _adaptive_replay_set(
+            requests, study=disabled_study, window=window
+        )
+        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_linear_ms_ttft_rule_scales_with_input_length(self) -> None:
+        slo = SloSpec.from_dict(
+            {
+                "target_pass_rate": 0.95,
+                "ttft_rule": {"kind": "linear_ms", "intercept_ms": 4000, "per_token_ms": 0.125},
+                "tpot_rule": {"kind": "fixed_ms", "threshold_ms": 50},
+            }
+        )
+
+        def ev(prompt_tokens: int, ttft_ms: float):
+            return evaluate_request(
+                RequestOutcome(
+                    request_id="r",
+                    success=True,
+                    ttft_ms=ttft_ms,
+                    tpot_ms=10.0,
+                    prompt_tokens=prompt_tokens,
+                    completion_tokens=8,
+                ),
+                slo,
+            )
+
+        # threshold = 4000 + 0.125*L_in : 8k->5000ms, 0->4000ms
+        self.assertTrue(ev(8000, 4900).passed)
+        self.assertFalse(ev(8000, 5100).passed)
+        self.assertTrue(ev(0, 3900).passed)
+        self.assertFalse(ev(0, 4100).passed)
+
+    def test_streaming_socket_timeout_is_a_failed_request_not_a_crash(self) -> None:
+        # A request that exceeds request_timeout_s raises TimeoutError mid-stream;
+        # it must surface as HttpClientError (a failed request), never escape to
+        # crash the trial.
+        with mock.patch(
+            "aituner.http_client._urlopen", side_effect=TimeoutError("timed out")
+        ):
+            with self.assertRaises(HttpClientError):
+                stream_chat_completion(
+                    base_url="http://127.0.0.1:1/v1",
+                    body={"messages": [{"role": "user", "content": "hi"}], "stream": True},
+                    timeout_s=0.5,
+                )
+            outcome = _run_one_request(
+                TraceRequest(
+                    row_id="r",
+                    arrival_s=0.0,
+                    sampling_u=1.0,
+                    body={"messages": [{"role": "user", "content": "hi"}], "stream": True},
+                    prompt_tokens_hint=10,
+                    completion_tokens_hint=None,
+                ),
+                base_url="http://127.0.0.1:1/v1",
+                timeout_s=0.5,
+            )
+        self.assertFalse(outcome.success)
+        self.assertIn("timed out", outcome.error)
+
+    def test_sigterm_is_converted_to_keyboardinterrupt(self) -> None:
+        # So a killed `study tune` runs the engine-teardown finally instead of
+        # orphaning the vLLM EngineCore workers on the GPUs.
+        import signal as _signal
+
+        previous = _install_sigterm_as_keyboardinterrupt()
+        try:
+            with self.assertRaises(KeyboardInterrupt):
+                _signal.raise_signal(_signal.SIGTERM)
+        finally:
+            _restore_sigterm(previous)
+
     def test_lca_similarity_matrix_separates_different_profiles(self) -> None:
         window = WindowRecord(
             window_id="base",
@@ -3796,6 +4101,56 @@ class CoreFlowTests(unittest.TestCase):
             state = store.load_state("study-1")
             self.assertEqual(state.next_trial_index, 1)
 
+    def test_cli_tune_vetoes_unauthorized_llm_stop(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            tmp_path = Path(tmp)
+            study_path = _write_study_assets(tmp_path)
+            spec = json.loads(study_path.read_text(encoding="utf-8"))
+            spec["llm"]["endpoint"] = {
+                "provider": "custom",
+                "base_url": "http://localhost:9/v1",
+                "model": "test-model",
+                "api_key_env": "AITUNER_TEST_KEY",
+            }
+            study_path.write_text(json.dumps(spec), encoding="utf-8")
+            store_root = tmp_path / "store"
+            stop_payload = json.dumps(
+                {
+                    "observation": "looks done",
+                    "diagnosis": "agent thinks it converged",
+                    "config_patch": {"env_patch": {}, "flag_patch": {}},
+                    "expected_effects": ["stop"],
+                    "why_not_previous_failures": "n/a",
+                    "should_stop": True,
+                }
+            )
+            buffer = io.StringIO()
+            with mock.patch("aituner.cli.run_trial") as run_trial_mock, mock.patch(
+                "aituner.cli.call_llm_for_proposal", return_value=stop_payload
+            ), contextlib.redirect_stdout(buffer):
+                exit_code = cli_main(
+                    [
+                        "study",
+                        "tune",
+                        "--spec",
+                        str(study_path),
+                        "--store-root",
+                        str(store_root),
+                        "--skip-baseline",
+                        "--max-trials",
+                        "2",
+                    ]
+                )
+            self.assertEqual(exit_code, 0)
+            run_trial_mock.assert_not_called()
+            executed = json.loads(buffer.getvalue())["executed_trials"]
+            # The first unauthorized LLM stop is vetoed; the second is honored
+            # only after the veto budget is spent.
+            self.assertTrue(any(item.get("stop_vetoed") for item in executed))
+            honored = [item for item in executed if item.get("stopped")]
+            self.assertTrue(honored)
+            self.assertEqual(honored[-1]["stop_authorized_by"], "llm_after_veto_budget")
+
     def test_cli_tune_uses_harness_stop_before_llm(self) -> None:
         with tempfile.TemporaryDirectory() as tmp:
             tmp_path = Path(tmp)

tests/test_prepare_trace_windows.py

@@ -0,0 +1,100 @@
+from __future__ import annotations
+
+import importlib.util
+import sys
+import unittest
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+_SPEC = importlib.util.spec_from_file_location(
+    "prepare_trace_windows",
+    REPO_ROOT / "scripts" / "prepare_trace_windows.py",
+)
+assert _SPEC and _SPEC.loader
+ptw = importlib.util.module_from_spec(_SPEC)
+# Register before exec so dataclasses can resolve the module's annotations.
+sys.modules[_SPEC.name] = ptw
+_SPEC.loader.exec_module(ptw)
+
+
+class SessionCoherentSamplingTests(unittest.TestCase):
+    def test_multi_hop_chain_resolves_to_root(self) -> None:
+        root_of: dict[object, object] = {}
+        # turn1 root, turn2 -> turn1, turn3 -> turn2 (multi-hop), streamed in order.
+        self.assertEqual(
+            ptw.resolve_session_root({"chat_id": 1, "parent_chat_id": -1, "turn": 1}, root_of),
+            1,
+        )
+        self.assertEqual(
+            ptw.resolve_session_root({"chat_id": 2, "parent_chat_id": 1, "turn": 2}, root_of),
+            1,
+        )
+        self.assertEqual(
+            ptw.resolve_session_root({"chat_id": 3, "parent_chat_id": 2, "turn": 3}, root_of),
+            1,
+        )
+
+    def test_unknown_parent_falls_back_to_parent_id(self) -> None:
+        root_of: dict[object, object] = {}
+        # parent never seen (fell outside the span): group siblings under the parent.
+        self.assertEqual(
+            ptw.resolve_session_root({"chat_id": 50, "parent_chat_id": 9, "turn": 2}, root_of),
+            9,
+        )
+        self.assertEqual(
+            ptw.resolve_session_root({"chat_id": 51, "parent_chat_id": 9, "turn": 2}, root_of),
+            9,
+        )
+
+    def test_all_turns_of_a_session_share_one_u(self) -> None:
+        root_of: dict[object, object] = {}
+        rows = [
+            {"chat_id": 1, "parent_chat_id": -1, "turn": 1},
+            {"chat_id": 2, "parent_chat_id": 1, "turn": 2},
+            {"chat_id": 3, "parent_chat_id": 2, "turn": 3},
+        ]
+        us = {
+            ptw.session_uniform(
+                seed=7,
+                window_id="w",
+                session_root=ptw.resolve_session_root(row, root_of),
+            )
+            for row in rows
+        }
+        self.assertEqual(len(us), 1)
+        only = next(iter(us))
+        self.assertGreaterEqual(only, 0.0)
+        self.assertLess(only, 1.0)
+
+    def test_thresholding_keeps_or_drops_whole_sessions(self) -> None:
+        # Two distinct sessions get distinct scores; a threshold either keeps a
+        # session's every turn or none of them.
+        seed, window_id = 20260325, "chat_w_x"
+        sessions = {
+            "A": [
+                {"chat_id": 10, "parent_chat_id": -1},
+                {"chat_id": 11, "parent_chat_id": 10},
+            ],
+            "B": [
+                {"chat_id": 20, "parent_chat_id": -1},
+                {"chat_id": 21, "parent_chat_id": 20},
+            ],
+        }
+        root_of: dict[object, object] = {}
+        scored: list[tuple[str, float]] = []
+        for name, rows in sessions.items():
+            for row in rows:
+                root = ptw.resolve_session_root(row, root_of)
+                u = ptw.session_uniform(seed=seed, window_id=window_id, session_root=root)
+                scored.append((name, u))
+        for name in sessions:
+            us = {u for n, u in scored if n == name}
+            self.assertEqual(len(us), 1, f"session {name} turns must share one u")
+        for threshold in (0.0, 0.25, 0.5, 0.75, 1.0):
+            for name in sessions:
+                kept = {u <= threshold for n, u in scored if n == name}
+                self.assertEqual(len(kept), 1, "a session must be kept/dropped as a whole")
+
+
+if __name__ == "__main__":
+    unittest.main()