Add knob conditional effect figures

docs/aituner-harness-design-contract.md

@@ -0,0 +1,333 @@
+# AITuner Harness Design Contract
+
+本文总结当前 AITuner harness 的设计语义。它不是实验流水账，也不是最终论文文字；
+它的作用是把我们能 claim 的系统贡献、各模块做法、隐含假设和限制说清楚。
+
+核心结论：
+
+```text
+AITuner harness 的贡献不是“LLM 会调参”，也不是“写了一组专家 if/else 规则”。
+
+Harness 的目标是把 black-box knob search 转成：
+measurement-grounded, mechanism-guided, validator-controlled experiments。
+```
+
+换句话说，planner 可以是 LLM、BO、bandit、deterministic heuristic 或人工选择。
+Harness 负责把观测转换成可审计的机制假设，生成合法候选，并用真实测量验证或否定这些假设。
+
+## 核心状态机
+
+```mermaid
+flowchart LR
+    S[State<br/>workload, constraints, history] --> E[Evidence<br/>SLO symptoms to mechanism signals]
+    E --> C[CandidateSet<br/>typed interventions]
+    C --> V[Validator<br/>legal, novel, covered?]
+    V -->|run trial| M[Measurement<br/>verdict]
+    M --> S
+    V -->|no justified candidate| X[Stop / report]
+```
+
+Harness 的核心循环只有五步：
+
+1. **State**：维护 workload、SLO、engine/hardware constraints 和历史 trial measurement。
+2. **Evidence**：把 probe 结果从 raw logs 转成 serving-stage symptom signals。
+3. **CandidateSet**：在 mechanism space 中生成有限个 typed interventions。
+4. **Validator**：检查 legality、full-config novelty、failure memory 和 coverage。
+5. **Measurement**：执行被验证过的 intervention，用真实 SLO verdict 更新状态；若没有
+   justified candidate，则 stop 或报告 measurement/coverage gap。
+
+这个状态机表达的是 harness 的最小设计，不依赖具体 planner。LLM、BO、bandit 或
+deterministic heuristic 都只能在 `CandidateSet` 上排序或选择，不能绕过 `Validator`
+直接构造 config，也不能单方面决定 stop。
+
+## 核心设计不变量
+
+后续所有低层模块都服务于三个不变量：
+
+| 不变量 | 含义 | 为什么重要 |
+| --- | --- | --- |
+| Measurement-grounded | 每个状态转移都由真实 probe/SLO verdict 更新 | 防止 planner 把自然语言猜测当成事实 |
+| Mechanism-typed | 候选不是裸 knob vector，而是 topology/scheduler/admission/cache 等 intervention | 降低搜索维度，并让每个 trial 有可解释假设 |
+| Validator-controlled | candidate 和 stop 必须通过 legality、no-repeat、coverage 和 failure guards | 防止重复实验、非法配置和 premature stop |
+
+## 从 High Level 到 Low Level 的展开
+
+下面各节按实现层次展开：
+
+1. Observation schema 定义 harness 能看到什么；
+2. Evidence compiler 说明 symptom 如何变成机制证据；
+3. Mechanism space 说明候选空间从哪里来；
+4. CandidateSet 说明如何构造 intervention；
+5. Planner interface 说明 LLM/BO/heuristic 的边界；
+6. Validator 说明什么能执行、什么能停止。
+
+每一层都区分两件事：当前 prototype 的具体做法，以及这些做法的假设和限制。
+
+## 详细模块语义
+
+### 1. Observation Schema
+
+Harness 先把一次 trial/probe 的结果转成结构化 observation：
+
+```text
+O_t = {
+  workload summary,
+  SLO rules,
+  effective engine config,
+  best feasible probe,
+  limiting probe,
+  failed_reason_counts,
+  early_stop_reason,
+  pass_rate,
+  request_rate_per_gpu,
+  launch / OOM status
+}
+```
+
+其中 `failed_reason_counts` 的定义是 request-level SLO violation reason 的 multiset 计数：
+
+```text
+ttft_ms>threshold       request 的 TTFT 超 SLO
+tpot_ms>threshold       request 的 TPOT 超 SLO
+arrival_lag_s>limit     synthetic arrivals 已经追不上
+probe_elapsed_s>limit   probe 总耗时超过上限
+slo_pass_rate_unrecoverable
+                        已失败过多，数学上无法达到 target pass rate
+request_failed / timeout / completion mismatch
+                        请求级失败
+```
+
+重要限制：
+
+- 一个 request 可以同时贡献 `ttft_ms>...` 和 `tpot_ms>...`；
+- `failed_reason_counts` 是 symptom evidence，不是 root-cause ground truth；
+- queueing/admission 主要来自 probe 调度层 early stop，而不是单个 request latency 的精确分解。
+
+因此文档和论文里必须避免说“failed count 证明 root cause 是 TTFT/TPOT”。更准确的说法是：
+
+```text
+failed_reason_counts gives SLO violation symptoms.
+Harness infers serving-stage hypotheses from these symptoms.
+```
+
+### 2. Evidence Compiler / Bottleneck Hypotheses
+
+当前 prototype 做两层聚合。
+
+第一层是单个 probe 的 active bottleneck。当前实现用 count-majority：
+
+```text
+ttft_count = sum(count(reason startswith "ttft"))
+tpot_count = sum(count(reason startswith "tpot"))
+other_request_failed_count = non-TTFT, non-TPOT request failures
+
+if ttft_count >= max(tpot_count, other_request_failed_count):
+    active = ttft_prefill
+elif tpot_count >= max(ttft_count, other_request_failed_count):
+    active = decode_tpot
+else:
+    active = admission_or_queueing
+```
+
+这一步是 heuristic。它的语义基础是 TTFT/TPOT/arrival-lag 对应不同 serving stages，
+但 “majority label = root cause” 并不成立。
+
+第二层是跨 trial 的 ranked hypotheses。它把以下证据合成 score：
+
+- workload prior：decode-only + TPOT SLO 更支持 decode hypothesis；长 prompt tail 更支持 prefill hypothesis；
+- latest probe active label；
+- historical probe evidence；
+- `failed_reason_counts` 中 TTFT/TPOT/queueing symptom ratio；
+- launch failure / OOM。
+
+更稳健的目标设计应该把第一层 hard label 改成 soft evidence vector：
+
+```text
+e_prefill   = normalized count of TTFT symptoms
+e_decode    = normalized count of TPOT symptoms
+e_admission = normalized count of arrival lag / elapsed / unrecoverable / request failures
+e_memory    = launch or OOM evidence
+```
+
+Candidate generator 应该基于 evidence distribution 生成 top mechanism probes，而不是只相信一个
+hard dominant bottleneck。当前 prototype 的 hard label 是工程近似，不是最终 contribution。
+
+### 3. Mechanism Space
+
+Harness 不在 raw knob Cartesian product 中盲搜：
+
+```text
+raw space = {TP, DP, EP, GMU, MNS, MBT, chunked-prefill, ...}
+```
+
+它先把 knobs 映射到 serving pipeline 上的可控 mechanism：
+
+| Mechanism family | Example knobs | 机制含义 | 典型 evidence |
+| --- | --- | --- | --- |
+| Topology / resource partition | TP, DP, EP, visible GPUs | 改变 compute/memory 分布、replica 数、per-GPU efficiency | TTFT/TPOT pressure, topology frontier 未覆盖 |
+| Prefill scheduler | chunked prefill, MBT | 改变 prefill quantum 和 head-of-line blocking | TTFT symptoms, long prompt tail, low prefix reuse |
+| Admission / concurrency | MNS | 改变活跃 sequence 数和 batch/admission pressure | arrival lag, pass-rate unrecoverable, concurrency underuse |
+| KV/cache headroom | GMU, block/cache knobs | 改变 KV cache blocks 和 memory feasibility | cache pressure, launch/memory, topology settled 后仍有 SLO pressure |
+| Launch/memory feasibility | env, memory-affecting flags | 确认 engine 是否能启动、是否 OOM | launch failure, OOM |
+| Frontier delta transfer | measured runtime delta applied to other Pareto anchors | 将已测 runtime 改动投影到未测 frontier anchor | 同 topology 上 runtime delta 为正，且存在其他 Pareto anchor |
+
+这些 family 的依据不是某个 case 的 winning config，而是 LLM serving pipeline：
+
+```text
+arrival/admission -> prefill -> decode -> memory/launch feasibility
+```
+
+每个 family 必须满足三条约束：
+
+1. 它对应一个可解释的 serving mechanism；
+2. 它只生成 engine schema 和 hardware constraints 下合法的 candidate；
+3. 它的 confirm/reject condition 由真实 measurement 决定。
+
+限制：
+
+- Mechanism family 是 domain-specific，不是 engine-agnostic magic；
+- vLLM/SGLang 等 engine 的 knob 名称不同，需要 adapter 把 engine knobs 映射到同一 mechanism vocabulary；
+- family 本身有系统依据，但当前 score 常数和部分 gate 仍是 heuristic。
+
+### 4. CandidateSet / Intervention Generation
+
+Candidate 不是“一个 patch”这么简单。一个合法 candidate 应包含：
+
+```text
+candidate = {
+  mechanism_family,
+  config_patch,
+  hypothesis,
+  expected_effects,
+  confirm_condition,
+  reject_condition,
+  effective_full_config_signature
+}
+```
+
+当前 prototype 的候选顺序大致是：
+
+1. topology candidates；
+2. frontier-delta projection candidates；
+3. runtime candidates。
+
+其中 runtime candidates 又包含 prefill scheduler、MBT、MNS、GMU 等 family。
+
+设计假设：
+
+- topology/resource partition 通常改变较大的 capacity frontier；
+- runtime knobs 通常是同一 topology 下的 local refinement；
+- 当 topology frontier 未覆盖时，过早 runtime hill-climbing 可能把搜索困在坏 topology；
+- 当一个 runtime delta 已在某个 topology 上测得正收益时，把这个 delta 投影到其他 Pareto
+  anchor 是比完整 factorial grid 更便宜的 interaction test。
+
+限制：
+
+- `topology-before-runtime` 是强 prior，不是定理；需要 ablation；
+- frontier delta transfer 依赖已测 history，如果 history 太少就不能工作；
+- 当前 prototype 中一些 target step 和 score 常数仍然是人工 heuristic。
+
+### 5. Planner Interface
+
+Planner 的职责应该被限制为：
+
+```text
+rank/select candidate from CandidateSet
+explain why this candidate is worth the next trial
+```
+
+Planner 不应该：
+
+- 构造 schema 外的 knob；
+- 绕过 topology / memory constraints；
+- 重复已经测试过的 effective full config；
+- 单方面决定 stop；
+- 把自然语言猜测当成 measurement verdict。
+
+这也是 no-LLM harness 能工作的原因：只要 `CandidateSet` 和 `Validator` 足够有信息，
+一个 deterministic planner 也可以完成 tuning。LLM 的价值在于组合 evidence、解释 tradeoff、
+在候选较多时排序，而不是提供 tuning correctness 的唯一来源。
+
+### 6. Validator / Stop Authority
+
+Validator 是 harness 防止 prompt engineering 化的关键。它负责：
+
+- canonicalize effective full config；
+- 拒绝 no-op 或 repeat；
+- 检查 legal topology / visible GPU / tunable schema；
+- 记录 failure memory；
+- 判断 measurement ceiling，例如 `search.high` 是否不足；
+- 在 candidate coverage 不足时禁止 premature stop；
+- 只有在覆盖和 measurement guards 都满足时授权 stop。
+
+重要设计修正：
+
+```text
+no-repeat must use normalized effective full-config signature,
+not patch signature.
+```
+
+因为 runtime-only patch 在 materialization 时会继承 incumbent topology。
+如果只看 patch signature，可能把 `{"gmu": 0.9}` 误认为新 config，
+但真实执行时它可能 materialize 成已测过的 full config。
+
+限制：
+
+- Validator 只能保证相对于声明的 grammar/operator set 的 coverage；
+- 它不能证明全 raw knob space 没有更优点；
+- measurement ceiling 不足时应报告并请求人类确认，而不是静默合成 arrivals 或重复窗口。
+
+## 精确贡献表述
+
+我们应该 claim：
+
+```text
+AITuner introduces a planner-agnostic harness that converts LLM serving
+configuration tuning from black-box knob search into typed, measurement-grounded
+counterfactual experiments over serving mechanisms.
+```
+
+可拆成三点贡献：
+
+1. **Serving-stage evidence compiler**
+   将 workload profile、SLO violation symptoms、probe early stop 和 launch failure
+   转换为 prefill/decode/admission/memory/launch 的机制证据，而不是只给 planner 一个 scalar score。
+
+2. **Typed mechanism action space**
+   将 raw knobs 组织为 topology、prefill scheduler、admission/concurrency、cache headroom、
+   frontier transfer 等 intervention families，使搜索发生在 mechanism space 而不是任意 knob vector space。
+
+3. **Validator-controlled experimental loop**
+   用 full-config signature、constraints、failure memory、coverage 和 measurement guards
+   控制 proposal 与 stop，使 LLM/BO/heuristic 都只能在合法、可审计的 candidate set 上工作。
+
+我们不应该 claim：
+
+- bottleneck classifier 永远正确；
+- `failed_reason_counts` 是 root cause label；
+- 当前 heuristic score 常数有理论最优性；
+- harness 覆盖完整 raw knob space；
+- stop 证明全局最优；
+- 某个 case 的 winning config 被系统“证明”出来。
+
+## 必须补的证据
+
+为了证明贡献不是 rule accumulation，后续实验必须 ablate family 和 authority，而不是只报最终性能：
+
+| Ablation | 证明什么 |
+| --- | --- |
+| classifier off / shuffled evidence | evidence attribution 是否真的影响正确方向 |
+| mechanism space off，改用 raw random/BO | mechanism action space 是否压缩搜索并提升收敛 |
+| topology-before-runtime off | 大 frontier intervention prior 是否必要 |
+| frontier-delta projection off | cross-topology runtime transfer 是否解决 bad-start/local trap |
+| validator off / patch signature only | full-config validator 是否避免重复和 false progress |
+| no-LLM deterministic planner | harness 是否是 planner-agnostic substrate |
+| weak planner + harness vs strong planner naive | harness 是否能补偿 planner 能力差距 |
+
+最终论文表达应保持这个边界：
+
+```text
+Harness makes the search more structured, auditable, and measurement-efficient.
+It does not replace measurement, does not prove global optimality, and does not
+turn symptom labels into perfect causal diagnosis.
+```

docs/aituner-harness-summary.md

@@ -1,5 +1,23 @@
 # AITuner Harness Summary
 
+## No-LLM Deterministic Planner
+
+当前 harness 不只是给 LLM 的 prompt hints。它已经可以在没有 LLM endpoint 的情况下，
+作为 deterministic planner 完成一整轮 tuning：
+
+1. 先运行 baseline，得到真实 probe/SLO evidence。
+2. 从 probe history 构造 trial profile 和 bottleneck hypotheses。
+3. 从 topology/runtime intervention grammar 中生成合法 candidate actions。
+4. 用 expected relief、information gain、launch safety 和 regression risk 给候选打分。
+5. 若高分候选存在，直接写出 `harness-proposal-XXXX`。
+6. 若候选耗尽，且 validator 证明 post-incumbent validation 已充分，写出
+   `harness-stop-XXXX`。
+7. 只有 harness 既不能 propose 也不能 stop 时，才调用 LLM；如果没有 LLM endpoint，
+   tune loop 会显式失败。
+
+完整机制和 Qwen30B no-LLM 真实轨迹见：
+[No-LLM harness mechanism](harness-ablation/no-llm-harness-mechanism-20260625.md)。
+
 ## What The Harness Adds
 
 The harness turns each LLM proposal from open-ended config search into a bottleneck-directed decision.

docs/aituner-maas-collab-overview-20260703.md

@@ -0,0 +1,221 @@
+# AITuner：MaaS Serving Config 自动调优 — 合作概述
+
+> 面向：配置调优团队
+> 目的：介绍 AITuner 是什么、为什么它比纯 LLM loop 可靠、以及我们建议的 pilot 合作方式。
+> 日期：2026-07-03
+
+## 一句话总结
+
+AITuner 把 LLM serving engine 的 config tuning 从"人工试错 / LLM 黑盒瞎猜"变成
+**基于真实测量、按 bottleneck 机制分类、由 validator 把关的自动实验循环**。
+我们希望在贵团队的真实环境上跑通 1-2 个 case，验证它能否成为你们的日常工具。
+
+## 1. 问题背景
+
+MaaS 场景下 tuning 的现状：
+
+- 平台上有**数百个模型**，模型本身还在持续迭代；
+- 硬件平台在更新，同一个模型在不同硬件上的最优 config 不同；
+- 每个 (model, hardware, workload, SLO) 组合都是一个独立的 tuning case；
+- 人力远远覆盖不了所有 case，大量 case 只能用默认或粗调的 config 上线，
+  留下吞吐和成本上的浪费。
+
+AITuner 的目标：**自动化 tune 这些没人力覆盖的 case**，输出满足 SLO 的
+engine config，并附带可审计的实验证据。
+
+## 2. 为什么"纯 LLM 自动调参"不够
+
+直接让 LLM 在循环里提 config、跑一轮、再提 config，实践中有两个硬伤：
+
+**(a) 缺 domain-specific 知识和内部 context。**
+
+- LLM 会误读 engine knob 的语义，典型例子是 vLLM 的 DP：LLM 常把它当成
+  "免费加吞吐"的开关，忽略它改变的是 replica 数和 per-GPU 效率，
+  在 per-GPU 指标下盲目 scale-out 反而变差；
+- 内部平台（如 dash）有大量内部环境变量和 launch 约束，LLM 完全没有这部分
+  context，提出的 config 经常直接 launch failure 或 OOM。
+
+**(b) 缺 bottleneck breakdown，不会像专家一样理解系统。**
+
+- LLM 拿到的往往只是"pass rate 低了"这样的 scalar 结果，
+  它无法区分瓶颈在 prefill、decode、admission/queueing 还是 memory；
+- 没有瓶颈归因，proposal 就退化成 knob space 里的随机游走：
+  重复已试过的配置、在错误的 knob family 上反复消耗 GPU trial。
+
+AITuner 的设计就是补上这两块：把系统知识和瓶颈分析放进一个 **harness**，
+LLM 只负责在 harness 给出的合法候选里做排序和取舍。
+
+## 3. AITuner 是怎么工作的（概述）
+
+一句话：AITuner 是一个自动实验循环——对目标 case 做真实压测，
+从结果做瓶颈归因，按系统机制生成下一个 config 候选，
+经 validator 把关后执行下一轮测量，直到证据表明继续实验不再值得。
+
+```text
+真实压测 ──> 瓶颈归因 ──> 机制化候选 ──> Validator 把关 ──> 下一轮压测 / 停止
+```
+
+对使用方来说，需要知道的只有四点：
+
+1. **每一步决策都来自真实测量**，不是 LLM 的自然语言猜测——每轮以
+   SLO verdict、pass rate、`request_rate_per_gpu`、launch/OOM 状态为准；
+2. **像专家一样先归因再动手**：把 SLO 违约症状聚合成 prefill / decode /
+   admission / memory 瓶颈假设，候选 config 只从对应的机制
+   （拓扑切分、prefill 调度、并发准入、KV cache 余量等）中生成，
+   不在 knob 空间里盲搜；
+3. **Validator 挡住不合法和重复的实验**：engine 参数合法性、硬件/拓扑约束、
+   内部平台 launch 约束、已测配置查重、失败记忆——包括 LLM 在内的任何
+   proposal 来源都必须过这一关；
+4. **知道什么时候该停**：验证充分或触及测量上限时确定性停止，
+   不多烧 GPU，也不会静默宣称"已经最优"。
+
+架构与模块细节见 `docs/aituner-harness-design-contract.md`，
+pilot 阶段可按需深入，这里不展开。
+
+## 4. 关键性质：不被单一 LLM 绑死
+
+Harness 是 **planner-agnostic** 的：LLM、确定性 heuristic 甚至 BO/bandit
+都只是在同一个 CandidateSet 上做排序。目前已经验证：
+
+- **No-LLM 模式**：在没有任何 LLM endpoint 的情况下，harness 可以作为
+  deterministic planner 完成整轮 tuning（baseline → 假设 → 候选 → 打分 →
+  proposal/stop），已有 Qwen30B 真实轨迹；
+- 高分确定性候选存在时根本不调 LLM，LLM 只在候选需要复杂 tradeoff
+  排序时介入。
+
+这意味着**换 LLM 供应商的风险是可控的**：tuning 的正确性来自
+harness 的证据编译和 validator，而不是某个特定模型的能力。
+
+## 5. 已有证据（内部实验，3 个 case）
+
+对照组均为"纯 LLM loop"（同一 LLM、同一压测框架，只关闭 harness），
+指标为满足 SLO 的 `request_rate_per_gpu`（每 GPU 可承载请求率，越高越好）。
+
+### Case 1：qwen27b chat 0-8k，dash0 内部 vLLM，H20
+
+真实 trace 窗口回放（`chat_w20260311_1000`），SLO：95% pass rate、
+TTFT 2s/4s/6s 分档、TPOT ≤ 50ms。
+详见 `docs/qwen27b-chat-0-8k-current-config-fig18-20260506.md`。
+
+| | 纯 LLM loop | AITuner |
+| --- | --- | --- |
+| 最终最优 config | TP2/DP1，**0.2025** req/s/GPU | TP4，**0.4429** req/s/GPU（**约 2.2x**） |
+| baseline（起点相同） | TP1/DP1，0.0350 | TP1/DP1，0.0350 |
+| 搜索路径 | 第 2/3 轮先选 DP2、DP4，per-GPU 吞吐反而回落；第 4 轮才到 TP2 | 瓶颈归因判定 TTFT/prefill 主导，第 2 轮直接 TP2（0.2142），第 4 轮 TP4（0.4429） |
+| tuning 开销 | 跑满 12 轮 GPU trial，其中第 5-12 轮全部是无可行点的 runtime probe（纯浪费） | 第 8 轮确定性 stop，实际执行 4 次 GPU trial，全程约 2.5 小时 |
+
+两个值得注意的点：
+
+- 纯 LLM loop 的前几轮正是第 2 节所说的 DP 误读实例——LLM 把 DP scale-out
+  当成免费吞吐，per-GPU 效率被稀释，绕了 3 轮弯路；
+- 单轮真实 trial（engine launch + 多个二分 probe）约 1 小时，跑满 12 轮
+  意味着 10 小时以上的 GPU 占用；AITuner 在拿到约 2.2x 的 config 的同时，
+  把整个 tuning 过程压到约 2.5 小时。
+
+### Case 2：qwen235b thinking prefill（大模型，TP4 baseline）
+
+详见 `docs/qwen235b-thinking-prefill-harness-20260427.md`。
+
+| | 纯 LLM loop | AITuner |
+| --- | --- | --- |
+| 最优 config | TP8，**0.3794** req/s/GPU，第 **10** 轮才找到 | TP8，**0.3863** req/s/GPU，第 **2** 轮即超过对照组 12 轮的最优值 |
+| 搜索路径 | 中途浪费在 DP2、EP4 等失败探索上 | 从 baseline 直接跳到 TP8/DP1，跳过对照组踩过的失败方向 |
+| tuning 开销 | 12 轮预算 | 到达最优的迭代数从 10 降到 2（**5x**） |
+
+大模型 case 上单轮 trial 更贵，少跑 8 轮的绝对 GPU 成本节省也更大。
+
+### Case 3：Qwen3-30B-A3B，社区 vLLM 0.20（非内部环境同样适用）
+
+详见 `docs/qwen30b-community-vllm020/harness-early-stop-ablation-20260502.md`。
+两个子实验：
+
+- **测量上限识别**（search 上限较低时）：默认 config 已触及压测搜索上限。
+  AITuner 只执行 1 次 GPU trial 就识别出"当前压测范围测不出更优配置"，
+  停止并明确报告原因；纯 LLM loop 把 12 轮预算烧完——第 2 轮 DP2 使
+  per-GPU 吞吐减半，第 3-12 轮**连续 10 轮 launch failure**。
+  净节省 11 轮 GPU 占用。
+- **放宽上限后的真实搜索**：AITuner 第 4 轮到达最优 config family
+  （TP2 + runtime 精调）并在第 5 轮 stop；纯 LLM loop 经历 EP launch
+  failure、不可行 DP probe 等弯路后第 7 轮才到同一 family（单次测值差
+  约 1.5%，在重复压测噪声内）。
+
+### 汇总
+
+| Case | config 性能（vs 纯 LLM loop 最优） | 到达最优迭代数 | 省下的无效 GPU trial |
+| --- | --- | --- | --- |
+| qwen27b 内部 vLLM | **约 2.2x**（0.4429 vs 0.2025） | 4 vs 4，但对照组随后 8 轮全部无效 | 8 轮 infeasible probe |
+| qwen235b thinking | 持平略优（0.3863 vs 0.3794） | **2 vs 10（5x）** | 8 轮失败/弱探索 |
+| qwen30b 社区 vLLM | 同一 config family（±1.5%） | 4 vs 7；上限场景 1 vs 12 | 最多 11 轮（含 10 轮 launch failure） |
+
+诚实说明：以上是有限 case 上的证据，qwen30b 子实验使用了有界压缩回放
+（固定输出长度）做收敛性测试，不等同生产 benchmark；结论是"更快收敛到
+同等或更好的 config、大幅减少无效 GPU trial"，不是全局最优性证明。
+这正是我们想通过 pilot 在你们的真实 case 上进一步验证的。
+
+## 6. 你们会得到什么
+
+对每个 case，AITuner 的产出不只是一个 config：
+
+1. **满足 SLO 的 engine config**（以 `request_rate_per_gpu` 为主要
+   跨拓扑指标）；
+2. **完整可审计的实验轨迹**：每个 trial 的假设、预期效果、真实测量、
+   confirm/reject 结论，以及 probe 级别的明细（`probe_details.jsonl`）；
+3. **瓶颈归因报告**：这个 case 的限制因素是 prefill、decode、admission
+   还是 memory，为什么；
+4. **明确的 stop 理由**：是候选耗尽、验证充分，还是 measurement 上限
+   （比如 search 范围）不够——不会静默糊弄。
+
+这些轨迹本身对你们的人工 tuning 经验沉淀也有价值。
+
+## 7. 建议的合作方式（pilot）
+
+我们目前缺少的是贵团队每个 case 的真实环境和硬件。建议直接在你们的
+环境上做，分四步：
+
+**Phase 0 — 选 case、对齐输入（约 1 周）**
+- 双方选定 1-2 个有代表性的 case；
+- 每个 case 需要：模型 + 硬件规格、SLO 定义（TTFT/TPOT/pass rate）、
+  workload trace 或可复现的流量描述、当前人工 config（作对照基线）；
+- 我们把 dash 平台的内部环境变量 / launch 约束接入 harness 的
+  launch-feasibility 层和 validator。
+
+**Phase 1 — 在你们环境跑通（1-2 周）**
+- 部署 AITuner，先跑 baseline 建立测量基线，再跑完整 tune loop；
+- 我们负责跑通和调试，你们提供环境访问和平台侧支持。
+
+**Phase 2 — 对比评估**
+- AITuner 结果 vs 你们的人工 config：SLO 达标情况、`request_rate_per_gpu`、
+  消耗的 GPU trial 数、时间成本；
+- 全部结果附实验轨迹，可复查。
+
+**Phase 3 — 决策**
+- 若 pilot 达标，讨论扩展到更多 case 的方式（接入流程、权限、
+  运行成本、维护分工）。
+
+## 8. 当前依赖与风险（如实说明）
+
+1. **LLM 依赖**：当前 planner 使用 gpt-5.5。计划切换到百炼
+   qwen3.7-max 的 dog-fooding API，并做同 case 的效果对比。
+   风险缓冲：harness 的 no-LLM deterministic 路径已经能独立完成
+   相当一部分 tuning（见第 4 节），planner 模型的能力差距被 harness
+   部分补偿，切换成本预计可控——但对比数据出来之前这是一个待验证项。
+2. **Engine 适配**：当前 mechanism families 主要针对 vLLM 的 knob
+   语义；SGLang 等其他 engine 需要一层 adapter 把 knobs 映射到相同的
+   mechanism vocabulary（架构上已预留，工作量取决于目标 engine）。
+3. **平台适配**：dash 内部环境变量和 launch 约束需要在 Phase 0 一次性
+   接入，之后由 failure memory 持续积累。
+4. **边界**：AITuner 保证的是结构化、可审计、测量高效的搜索；
+   它不证明全局最优，瓶颈分类是 symptom-based 的启发式归因而非
+   完美因果诊断。对生产决策来说，可审计比"号称最优"更重要。
+
+## 9. 我们需要贵团队提供的
+
+- 1-2 个 case 的测试环境和硬件访问（或由你们的同学代跑，我们远程支持）；
+- 每个 case 的 SLO 定义和 workload trace；
+- dash 平台内部环境变量 / launch 约束的文档或对接人；
+- 百炼 qwen3.7-max dog-fooding API 的配额（用于 LLM 切换对比）。
+
+---
+
+附：更完整的设计语义见 `docs/aituner-harness-design-contract.md`，
+harness 各机制与实验证据见 `docs/aituner-harness-summary.md`。

docs/aituner-roadmap.md

@@ -39,6 +39,10 @@ M: measurement/scoring protocol
    SLO-constrained feasible frontier, req/s/GPU, latency quantiles, pass-rate guard
 ```
 
+当前 `src/aituner/harness.py` 是 prototype：它已经展示 no-LLM loop 和 mechanism-guided
+proposal 的可行性，但仍然包含大量 rule-based heuristics，不能作为最终 harness
+contribution。新的目标设计见 [Declarative intervention harness design](harness-ablation/declarative-intervention-harness-design-20260626.md)。
+
 Planner 是可替换的：
 
 ```text
@@ -60,6 +64,10 @@ kernel、KV cache、通信和排队的闭式性能模型。更稳妥也更强的
 
 ## 关键设计点
 
+当前 harness 设计语义、模块假设和限制见
+[AITuner Harness Design Contract](aituner-harness-design-contract.md)。Roadmap 只维护
+claim 和实验优先级；design contract 负责精确定义我们能说什么、不能说什么。
+
 | 设计点 | 更强表述 | 作用 | 需要证明 |
 | --- | --- | --- | --- |
 | Observation | mechanism state | 将 workload shape、probe trace、SLO failure、launch/memory failure 结构化 | agent 看到的是可计算状态，不是自然语言日志 |
@@ -72,17 +80,77 @@ kernel、KV cache、通信和排队的闭式性能模型。更稳妥也更强的
 
 | Claim | 当前状态 | 证据文档 | 关键缺口 |
 | --- | --- | --- | --- |
-| C1. Harness 将 raw knob search 转成 mechanism-guided intervention search，提升固定预算优化效果 | 已有强信号 | [Qwen27B 2x2](harness-ablation/qwen27b-tight-2x2-model-ablation-20260623.md), [Qwen30B SLO robustness](harness-ablation/qwen30b-slo-robustness-20260624.md) | 补 Qwen235B decode 2x2 aggregate；补 mechanism ablation |
+| C1. Harness 将 raw knob search 转成 mechanism-guided intervention search，提升固定预算优化效果 | 已有强信号 | [No-LLM harness mechanism](harness-ablation/no-llm-harness-mechanism-20260625.md), [Qwen27B 2x2](harness-ablation/qwen27b-tight-2x2-model-ablation-20260623.md), [Qwen30B SLO robustness](harness-ablation/qwen30b-slo-robustness-20260624.md) | 补 Qwen235B decode 2x2 aggregate；补 mechanism ablation |
-| C2. 收益来自 harness-defined substrate，不依赖某个强 LLM | 部分已有 | [Qwen27B 2x2](harness-ablation/qwen27b-tight-2x2-model-ablation-20260623.md) | 做 `BO/heuristic + harness` vs `BO/heuristic + raw knobs` |
+| C2. 收益来自 harness-defined substrate，不依赖某个强 LLM | Qwen30B no-LLM 已完整闭环；Qwen27B 弱/强模型已有 | [No-LLM harness mechanism](harness-ablation/no-llm-harness-mechanism-20260625.md), [Qwen27B 2x2](harness-ablation/qwen27b-tight-2x2-model-ablation-20260623.md) | 做 `BO/heuristic + harness` vs `BO/heuristic + raw knobs` |
 | C3. Weak planner + harness 可以匹配或超过 strong LLM naive | Qwen27B 已支持；Qwen235B 正在补 | [Qwen27B 2x2](harness-ablation/qwen27b-tight-2x2-model-ablation-20260623.md), [Qwen235B prefill progress](harness-ablation/qwen235b-prefill-2x2-progress-20260623.md) | 完成 Qwen235B decode 2x2；更新 prefill final doc |
-| C4. Attribution 和 intervention grammar 有机制贡献，不只是 prompt 信息更多 | 设计已有，严格证据不足 | [AITuner summary](aituner-harness-summary.md) | 做 shuffled attribution / no attribution / no grammar / no topology-first / no validator ablation |
+| C4. Attribution 和 intervention grammar 有机制贡献，不只是 prompt 信息更多 | 设计和 no-LLM case 已整理；严格 ablation 不足 | [No-LLM harness mechanism](harness-ablation/no-llm-harness-mechanism-20260625.md), [AITuner summary](aituner-harness-summary.md) | 做 shuffled attribution / no attribution / no grammar / no topology-first / no validator ablation |
 | C5. AITuner 找到 near-optimal region，而不是只找到一个可行 config | Qwen30B 有解释性信号 | [Qwen30B SLO robustness](harness-ablation/qwen30b-slo-robustness-20260624.md) | 选 1-2 个 case 做局部 grid 或专家配置对照 |
 | C6. AITuner 能随 SLO tightness 移动到合适 frontier | Qwen30B 已完成 | [Qwen30B SLO robustness](harness-ablation/qwen30b-slo-robustness-20260624.md) | 再选一个非同质 case 做 SLO sweep；同时画 SLO tightness -> frontier/regime transition |
 | C7. Engine adapter 让 intervention grammar 可迁移到其他 serving engine | 设计上可行，暂不作为主实验 claim | `EngineLaunchSpec` / launch recipe / tunable schema | vLLM 主线完成后，再做 SGLang adapter 和一个低成本验证 case |
+| C8. Harness 对坏初始点有恢复能力，不只依赖可信 base config | 单个 adversarial bad-start 已通过 first repair；分布级 robustness 不能 claim | [Declarative intervention harness design](harness-ablation/declarative-intervention-harness-design-20260626.md), [Bad-start stop counterexample](harness-ablation/bad-start-stop-counterexample-20260626.md), [Bad-start robustness suite](harness-ablation/bad-start-robustness-suite-20260626.md), [No-LLM harness mechanism](harness-ablation/no-llm-harness-mechanism-20260625.md) | 按 pre-registered 20-case suite 跑 random/adversarial start distribution |
 
 ## 最高优先级实验
 
-### P0. 完成 Qwen235B decode 2x2 并整理 aggregate
+### P0a. Declarative harness redesign gate
+
+目的：停止继续向 `harness.py` 添加 testcase-specific rules，把 harness 重构成
+declarative intervention grammar + coverage-relative validator。
+
+最低验收：
+
+- CandidateSet 完整枚举并持久化 snapshot；
+- CandidateSet v1 先限定为当前 harness generator 实际构造出的 concrete candidates，
+  不 claim 全 Cartesian knob space 枚举；`candidate_set_hash`、eligible/blocked
+  records 和 blocked reason summary 已在 harness context 与 `harness/candidate-set-*.json`
+  sidecar 中实现；
+- `harness_priority` 与 backend ranking 分离；
+- CoverageUnit 结构化，stop 不能只依赖 exact signature；
+- `search_high_saturated_by_incumbent` 不能绕过 CandidateSet coverage；对 `req/s/GPU`
+  目标，未覆盖 topology/resource-efficiency contrast 时必须继续；
+- 加入 `auto_search_high` measurement policy：可在已有 trace 内自动提高 ceiling；若
+  `search.high=1.0` 仍然不足，必须报告 `measurement_ceiling_insufficient` 并等待人类
+  确认，不得静默重复窗口或合成 arrivals；
+- normalized full-config signature：no-repeat 不能只看 patch signature；base config 与
+  no-op patch 必须被识别为同一 full config；`48911b6` 已实现并在 dash1 bad-start
+  validation 中通过；
+- materialized effective signature：runtime-only proposal 必须先按真实执行路径继承
+  incumbent topology，再做 no-repeat；已加入 shared signature/canonicalization，并在
+  CLI 进入 trial 前 hard-veto 重复 LLM/manual/harness proposal；
+- Failure invalidation 有保守 region predicate 和 retry/unblock 条件；
+- grammar/policy/capability 都有 version 和 anti-overfitting static checks；
+- LLM/BO 只能选择合法 candidate，不能绕过 validator。
+
+优先级原因：如果不先完成这个 gate，继续扩展 bad-start/SLO/2x2 实验只是在证明一个
+rule-based prototype。
+
+### P0b. Bad-start recovery confirmation after redesign
+
+目的：回答 harness 是否只能从可信 base config 起步，还是能从明显不合理的初始 config
+恢复到正确方向。
+
+Pre-registered suite 见 [Bad-start robustness suite](harness-ablation/bad-start-robustness-suite-20260626.md)。
+
+最小实验矩阵：
+
+| Case | 初始配置 | 证明点 |
+| --- | --- | --- |
+| bad-topology | `TP=8, DP=1` | 高 TP 起点会先做相邻低 TP bracket |
+| bad-runtime | `TP=2, gmu=0.5, max-num-seqs=8` | 低 runtime headroom 会跳回 nominal floor |
+| combined-bad | `TP=8, gmu=0.5, max-num-seqs=8` | topology recovery 和 runtime recovery 能串联 |
+
+注意：这不是先跑一条手工 bad case。必须在 declarative harness 之后跑 random/adversarial
+start distribution，并报告分布结果。
+
+预期图：
+
+- x-axis: trial index；
+- y-axis: best-so-far SLO-constrained req/s/GPU；
+- line groups: trusted-start vs bad-start cases；
+- annotation: proposal family sequence，例如 `TP downshift`, `gmu floor jump`, `gmu climb`。
+
+启动条件：先完成 P0a；再确认 dash fleet 有空闲 8xH20 机器；用户确认后再开跑。
+
+### P0c. 完成 Qwen235B decode 2x2 并整理 aggregate
 
 目的：补齐最核心的 `harness on/off x strong/weak planner` 证据，回答：
 

docs/harness-ablation/bad-start-robustness-suite-20260626.md

@@ -0,0 +1,122 @@
+# Bad-start robustness suite - 2026-06-26
+
+本文定义 P0 bad-start robustness 的分布级验证。它不是新的 claim 结果，而是下一轮实验的
+pre-registration：先固定 starts、指标和 pass/fail，再运行，避免根据单个 case 调规则。
+
+## 当前前提
+
+已完成的代码 gate：
+
+- normalized full-config signature；
+- materialized effective signature：runtime-only proposal 先继承 incumbent topology 再签名；
+- CLI hard-veto：LLM/manual/harness proposal 在进入 trial 前禁止重复 effective config；
+- CandidateSet audit：`candidate_set_hash`、eligible/blocked candidates、blocked reason summary；
+- sidecar persistence：`harness/candidate-set-*.json`。
+
+已通过的单 case：
+
+```text
+TP8, DP1, gmu0.5, max-num-seqs8
+  -> TP4
+  -> TP4 + gmu0.9
+```
+
+这个 case 只能证明 sentinel recovery，不能证明分布级 robustness。
+
+## 实验矩阵
+
+使用同一 Qwen30B-A3B community vLLM 0.20 bounded replay setup、no-LLM harness、
+`search.auto_high.enabled=true`。先跑 fresh trusted-start control，得到同 commit 下的
+参考值 `R_ref`。
+
+| Group | N | Initial starts | 证明点 |
+| --- | ---: | --- | --- |
+| trusted control | 1 | 可信/default start | 定义 `R_ref` |
+| topology-only | 4 | `(TP,DP)=(8,1),(4,2),(1,4),(2,4)`，runtime nominal | 证明不是只会 `TP8 -> TP4` |
+| runtime-only | 4 | `TP2/DP1` with `gmu={0.50,0.70}` and `max-num-seqs={8,16}` | 证明 runtime floor/admission recovery |
+| combined | 4 | `TP8/gmu0.70/mns16`, `TP4/DP2/gmu0.50/mns8`, `TP1/DP8/gmu0.50/mns16`, `TP2/DP4/gmu0.70/mns8` | 证明 operators 可串联 |
+| held-out random | 8 | fixed-seed stratified samples over legal topology x low/nominal `gmu` x low/normal `mns`，排除已通过 sentinel | overfit denominator |
+
+总计：1 control + 20 novel bad starts。
+
+## Primary metrics
+
+- best-so-far SLO-feasible `req/s/GPU / R_ref`；
+- time-to-95%-reference；
+- normalized AUC over trial budget；
+- final pass rate；
+- executed normalized full-config repeat count；
+- no-op blocked count；
+- candidate family / operator sequence；
+- stop reason and `candidate_set_hash`。
+
+每个 run 必须保留：
+
+```text
+state.json
+proposals/*.json
+harness/candidate-set-*.json
+trials/trial-*/trial_spec.json
+trials/trial-*/result.json
+```
+
+## Pass/fail
+
+Run-level pass：
+
+```text
+best_so_far_req_per_gpu >= 0.95 * R_ref within 12 measured trials
+pass_rate >= 0.95
+executed_effective_config_repeat_count == 0
+no harness stop while high-priority eligible candidates remain
+```
+
+Suite-level pass：
+
+```text
+20 / 20 novel bad starts pass.
+```
+
+如果任一 novel start 失败，不能 claim distribution-level bad-start robustness。修复后必须
+冻结失败分析，并重新抽 held-out random set。
+
+## Overfit guards
+
+- pre-register all starts and random seed；
+- 不把已通过的 exact `TP8,gmu0.5,mns8` sentinel 计入 20-case denominator；
+- 不在 starts 之间调 threshold；
+- 报告 operator names，例如 `topology_bracket`, `topology_redistribute`,
+  `runtime_floor_jump`, `admission_recovery`，而不是 case-specific action；
+- 每次 stop 必须引用 `candidate_set_hash` 和 no high-priority eligible candidate evidence。
+
+## GPU cost
+
+Expected：
+
+```text
+21 runs * 6-8 measured trials = 126-168 trials
+```
+
+Hard cap：
+
+```text
+21 runs * 12 measured trials = 252 trials
+```
+
+按当前 Qwen30B bounded replay 粗估：
+
+```text
+15-35 min / measured trial
+expected = 250-780 H20 GPU-hours
+cap      = 500-1175 H20 GPU-hours
+```
+
+因此建议先跑 3-case pilot：
+
+| Pilot case | 起点 | 目的 |
+| --- | --- | --- |
+| topology-only | `TP=4, DP=2, gmu=0.9, mns=64` | 检查不是只会处理 TP8 |
+| runtime-only | `TP=2, DP=1, gmu=0.5, mns=8` | 检查 runtime floor/admission recovery |
+| combined | `TP=1, DP=8, gmu=0.5, mns=16` | 检查 topology + runtime 串联 |
+
+Pilot 通过后再启动完整 20-case suite。

docs/harness-ablation/bad-start-stop-counterexample-20260626.md

@@ -0,0 +1,372 @@
+# Bad-start stop counterexample - 2026-06-26
+
+本文记录一次有意构造的 adversarial bad-start 测试。它的目的不是证明 harness 已经
+robust，而是攻击当前实现，确认它是否会从明显不合理的初始配置中恢复。
+
+结论：
+
+```text
+当前 production/prototype harness 还不能支持 bad-start robustness claim。
+
+它会在高 GPU、高 TP 的坏起点上被 search_high_saturated_by_incumbent 提前 stop，
+没有测试 topology/resource-efficiency contrast。
+```
+
+这不是一个需要补 `TP=8 -> TP=4` 特例规则的问题。它暴露的是更基础的 stop authority
+问题：measurement saturation 不能绕过 coverage-relative candidate set。
+
+同时，这个反例也暴露了 measurement policy 的缺口：`search.high` 太小时，tuning 会被
+offered-load ceiling 右截断。后续应该加入 `auto_search_high`，但它只能在已有 trace
+sampling space 内自动校准；如果 `search.high=1.0` 仍然不能压到真实 capacity frontier，
+系统必须主动报告 measurement ceiling 不足，并等待人类确认是否更换 trace、提高 trace
+density 或启用额外负载生成方式。
+
+## 实验设置
+
+机器：`dash1`，8x H20。
+
+目标：从一个故意不合理的初始配置开始：
+
+```text
+tensor-parallel-size = 8
+data-parallel-size = 1
+gpu-memory-utilization = 0.5
+max-num-seqs = 8
+LLM endpoint disabled
+```
+
+期望行为：
+
+- harness 不应只因为 baseline feasible 就停止；
+- 它至少应生成 topology/resource-efficiency contrast candidate；
+- 对 `req/s/GPU` 目标，8 GPU incumbent 需要被低 GPU 或邻域 topology probe 验证。
+
+## Run A: 低 search.high
+
+第一轮保留原始 `search.high=0.125`。
+
+结果：
+
+```text
+trial-0001 completed
+harness-stop-0002
+tuning_stop_reason = harness_stop
+validator reason = search_high_saturated_by_incumbent
+best request_rate = 1.0333 total
+best request_rate_per_gpu = 0.1292
+pass_rate = 1.0
+```
+
+解释：这个 run 的 offered-load ceiling 太低，baseline 很容易 saturate `search.high`。
+因此它不能区分“配置真的足够好”和“测量上限太低”。
+
+## Run B: corrected high search ceiling
+
+第二轮把 `search.high` 提到 `1.0`，保留同一个 bad-start 配置，`max_trials=3`。
+
+远端产物：
+
+```text
+session = adv_badcase_corr_casea_20260626T095356Z
+store = /home/admin/cpfs/wjh/aituner/aituner/.aituner/adversarial-badcase-corrected-casea-20260626T095356Z
+spec = /home/admin/cpfs/wjh/aituner/aituner/.aituner-run-configs/adversarial-badcase-corrected-casea-20260626T095356Z/casea-combined-bad-highsearch.json
+log = /home/admin/cpfs/wjh/aituner/aituner/.aituner/adversarial-badcase-corrected-casea-20260626T095356Z.log
+```
+
+结果仍然是在 baseline 后 stop：
+
+```text
+trial-0001 completed
+harness-stop-0002
+no harness-proposal-0002.json
+tuning_stop_reason = harness_stop
+validator reason = search_high_saturated_by_incumbent
+best sampling_u = 0.9375
+best request_rate = 8.033333333333333
+best request_rate_per_gpu = 1.0041666666666667
+pass_rate = 1.0
+```
+
+Probe trace：
+
+| sampling_u | request_rate | feasible |
+| --- | ---: | --- |
+| 0.5 | 4.6000 | true |
+| 0.75 | 6.5167 | true |
+| 0.875 | 7.5000 | true |
+| 0.9375 | 8.0333 | true |
+
+它触发 stop 的原因是当前 guard 计算：
+
+```text
+binary_probe_resolution = max(tolerance, (high - low) / 2**max_probes)
+                        = 0.0625
+threshold_gap_to_high = 1.0 - 0.9375
+                      = 0.0625
+```
+
+因此当前实现认为 incumbent 已经 saturate `search.high`。
+
+## 为什么这是反例
+
+当前 objective 是 SLO-constrained `req/s/GPU`，不是固定 8 GPU 的 total throughput。
+一个 8-GPU incumbent saturate offered-load ceiling，并不能证明：
+
+- 低 TP / 低 GPU 配置没有更高 `req/s/GPU`；
+- 当前 topology 是资源效率最优；
+- runtime knobs 已经进入合适 trust region；
+- no-LLM harness 能从 bad start 中恢复。
+
+所以这个 stop 是 unsound 的，至少相对于 bad-start robustness claim 是 unsound。
+
+更形式化地说：
+
+```text
+search_high_saturated_by_incumbent
+  does not imply
+incumbent_validated(topology/resource-efficiency)
+```
+
+当目标包含 resource efficiency，并且 parallel-size/topology 仍然 tunable 时，
+`search_high_saturated_by_incumbent` 只能作为 measurement evidence，不能单独作为 stop
+authority。
+
+## 对新 harness 设计的约束
+
+这个反例直接约束 declarative harness：
+
+1. Stop 前必须生成并持久化完整 `CandidateSet`。
+2. Stop proof 必须引用 `candidate_set_hash`。
+3. 如果存在未覆盖的 high-priority topology/resource-efficiency candidate，validator
+   必须返回 `eligible_candidates_remain`，即使 incumbent saturate `search.high`。
+4. `search.high` saturation 只能更新 measurement coverage，不能替代
+   `incumbent_validated`。
+5. 对 `req/s/GPU` objective，required coverage 必须包含至少一个 topology 或
+   resource-efficiency contrast，除非 StudySpec 明确固定 GPU budget 和 topology。
+
+Measurement policy 约束：
+
+1. `auto_search_high` 可以根据 trace 的 sampling threshold 和目标 GPU 规模自动提高
+   `search.high`，避免低 ceiling 让所有 config 过早 saturate。
+2. 自动校准不能越过 trace 原生上限。当前 `sampling_u` 语义下，`search.high=1.0`
+   表示完整 trace。
+3. 如果完整 trace 仍然被 incumbent 轻松 saturate，validator 不能假装搜索完成；它应该
+   输出 `measurement_ceiling_insufficient` 或把该事实作为 stop proof 的阻塞项。
+4. 系统不得自动使用重复窗口、合成 arrivals 或 replay scaling 来扩大 workload，除非
+   StudySpec 显式启用，或人类确认该实验要测 synthetic/offline stress regime。
+5. `measurement_ceiling_insufficient` 和 `eligible_candidates_remain` 是不同问题：前者说
+   load ceiling 不足，后者说 mechanism coverage 未完成。二者任一存在，都不能把结果
+   写成 bad-start robustness 成功。
+
+这也说明当前 repair 方向不能是：
+
+```text
+if tp == 8 and gmu == 0.5: try tp = 4
+```
+
+正确方向应该是：
+
+```text
+ordered topology lattice + resource-efficiency objective
+  -> candidate set includes lower/redistributed topology contrast
+  -> stop is blocked until that coverage unit is measured or invalidated
+```
+
+## 当前 verdict
+
+当前 production harness：
+
+```text
+prototype, not yet fundamental
+```
+
+新的 declarative prototype：
+
+```text
+promising substrate, but not production-proven
+```
+
+它已经把 `CandidateSet`、`CoverageUnit`、failure region 和 coverage-relative stop 的最小
+接口跑通，但还没接入真实 tuning loop，也还没证明 bad-start distribution 的收敛。
+
+因此接下来的 P0 gate 是：
+
+```text
+先实现 coverage-relative stop authority，再重跑 bad-start distribution。
+```
+
+## 2026-06-26 implementation validation
+
+Commit `c8a0f98` 实现了第一片 production 修复：
+
+- `search.auto_high` schema，默认关闭，旧配置兼容；
+- trial materialization 时在已有 trace sampling space 内 resolve effective `search.high`；
+- `trial_spec.json` 和 `result.json` 写入 auto-high / measurement evidence；
+- `search_high_saturated_by_incumbent` 降级为 measurement evidence；
+- 对 `req/s/GPU` 且 topology 可变的 study，high saturation 不能直接授权 stop；
+- 固定 GPU product 但 TP/DP redistribution 可调时，仍视为 topology 可变；
+- auto-high ceiling 低于 `search.low` 时不生成非法 search interval。
+
+本地验证：
+
+```text
+PYTHONPATH=src python3 -m unittest discover -s tests
+Ran 143 tests OK
+```
+
+dash1 validation：
+
+```text
+run label = adversarial-badstart-autohigh-c8a0f98-20260626T122622Z
+git sha   = c8a0f9870eac5438fb19be8edf1534a893723ab9
+machine   = dash1, 8x H20
+```
+
+Spec 仍使用 bad-start：
+
+```text
+tensor-parallel-size = 8
+data-parallel-size = 1
+gpu-memory-utilization = 0.5
+max-num-seqs = 8
+search.auto_high.enabled = true
+```
+
+Auto-high resolution：
+
+```text
+original_high       = 1.0
+effective_high      = 0.9979913161468553
+trace_max_sampling_u = 0.9979913161468553
+reason              = search_high_lowered_to_trace_ceiling
+```
+
+结果：
+
+| trial | config patch | best sampling_u | request_rate | req/s/GPU | pass |
+| --- | --- | ---: | ---: | ---: | ---: |
+| trial-0001 | baseline TP8, DP1, gmu0.5, mns8 | 0.935616858887 | 8.00 | 1.0000 | 1.0000 |
+| trial-0002 | `tensor-parallel-size=4` | 0.810867944369 | 6.95 | 1.7375 | 0.9784 |
+| trial-0003 | `tensor-parallel-size=8` | 0.935616858887 | 8.00 | 1.0000 | 1.0000 |
+
+关键结论：
+
+```text
+旧 failure 已被修复：
+baseline 后不再产生 harness-stop-0002/search_high_saturated_by_incumbent。
+
+新实现产生 harness-proposal-0002，并测试 TP4 topology contrast。
+TP4 将 best req/s/GPU 从 1.0000 提高到 1.7375。
+```
+
+这证明第一片修复解决了“measurement saturation 绕过 topology coverage”的问题。
+
+但是 trial-0003 暴露了新 blocker：
+
+```text
+当前 no-repeat 仍基于 patch signature，而不是 normalized full-config signature。
+```
+
+`tensor-parallel-size=8` 对这个 study 的 base config 是 no-op，等价于 baseline TP8，
+但系统仍把它当成一个新 proposal 执行。这说明下一片 P0 必须实现：
+
+1. normalized full-config signature；
+2. CandidateSet snapshot，包含 eligible 和 blocked candidates；
+3. blocked reason，例如 `blocked_noop_equivalent_to_tested_full_config`；
+4. Stop/report 中同时呈现 `measurement_ceiling_*` 和 `eligible_candidates_remain`。
+
+因此当前 verdict 更新为：
+
+```text
+P0 measurement/stop-order slice: passed.
+P0 full coverage-relative harness: not yet passed.
+```
+
+## 2026-06-26 normalized full-config validation
+
+Commit `48911b6` 修复了上一节暴露的新 blocker：no-repeat 不再只比较 patch
+signature，而是比较 normalized effective full config。
+
+实现语义：
+
+```text
+effective_config =
+  normalize(base_envs + env_patch,
+            base_flags + flag_patch)
+
+no_repeat_signature = stable_json(effective_config)
+```
+
+因此下面两个 proposal 在 validator 看来是同一个 full config：
+
+```text
+baseline patch: {}
+noop patch:    {"tensor-parallel-size": 8}
+```
+
+本地验证：
+
+```text
+PYTHONPATH=src python3 -m unittest discover -s tests
+Ran 145 tests OK
+```
+
+dash1 validation：
+
+```text
+run label = adversarial-badstart-fullsig-48911b6-20260626T133112Z
+git sha   = 48911b658bbf052d70d952d1cdf55ad6b50ba7a5
+machine   = dash1, 8x H20
+```
+
+Spec 仍使用同一个 adversarial bad-start：
+
+```text
+tensor-parallel-size = 8
+data-parallel-size = 1
+gpu-memory-utilization = 0.5
+max-num-seqs = 8
+search.auto_high.enabled = true
+LLM endpoint disabled
+```
+
+结果：
+
+| trial | proposal | best sampling_u | request_rate | req/s/GPU | pass |
+| --- | --- | ---: | ---: | ---: | ---: |
+| trial-0001 | baseline TP8, DP1, gmu0.5, mns8 | 0.935616858887 | 8.00 | 1.0000 | 1.0000 |
+| trial-0002 | `tensor-parallel-size=4` | 0.810867944369 | 6.95 | 1.7375 | 0.9832 |
+| trial-0003 | `tensor-parallel-size=4`, `gpu-memory-utilization=0.9` | 0.935616858887 | 8.00 | 2.0000 | 1.0000 |
+
+关键 observation：
+
+```text
+旧 trial-0003:
+  {"tensor-parallel-size": 8}
+  -> 等价于 baseline，但仍被执行
+
+新 trial-0003:
+  {"tensor-parallel-size": 4, "gpu-memory-utilization": 0.9}
+  -> 在已验证 TP4 topology 上继续测试 KV/cache headroom
+```
+
+这证明 normalized full-config signature 已经阻止了 patch-level no-op 重测。
+
+机制解释：
+
+1. baseline TP8 saturate search ceiling 只被记录为 measurement evidence；
+2. 因为 objective 是 `req/s/GPU`，topology/resource-efficiency contrast 仍未覆盖，所以
+   validator 不允许 stop；
+3. harness 先测试相邻低 TP topology，TP4 把 `req/s/GPU` 从 `1.0` 提高到 `1.7375`；
+4. no-repeat 用 full config signature block 掉等价 TP8 patch；
+5. harness 在 settled TP4 topology 上继续测试 runtime headroom，`gmu=0.9` 把
+   `req/s/GPU` 提高到 `2.0`。
+
+当前 verdict 更新为：
+
+```text
+P0 measurement/stop-order slice: passed.
+P0 normalized full-config no-repeat slice: passed.
+P0 single adversarial bad-start recovery: passed for this case.
+P0 distribution-level bad-start robustness: not yet proven.
+```

docs/harness-ablation/candidate-family-gap-log.md

@@ -0,0 +1,37 @@
+# Candidate Family Gap Review Log
+
+本文档维护 LLM 在 `advisory` 模式下提出 harness candidate set 之外配置、且该配置带来性能提升时的人工 review 入口。
+
+运行时系统不会自动修改 harness，也不会把 LLM 的 out-of-set proposal 直接提升为规则。每条提升先写入 study artifact：
+
+```text
+.aituner/<study>/candidate_family_gaps/<trial-id>.json
+```
+
+然后人工 review 决定是否需要修改：
+
+- `KnobDescriptor`
+- generic operator
+- acquisition / step-size policy
+- evidence estimator
+
+## Gap 分类
+
+| 类型 | 含义 | 默认处理 |
+|---|---|---|
+| `same_operator_new_step` | harness 已有同 knob / 同方向候选，但 LLM 给了更好的 step/value | 优先检查 trust-region、step-size、candidate budget 和 acquisition |
+| `missing_operator` | visible candidate set 中没有同 knob 或同 mechanism 的候选 | 检查是否缺 generic operator 或 descriptor 映射 |
+| `missing_descriptor` | knob 不在 adapter descriptor 中 | 需要 engine adapter review |
+| `missing_mechanism` | 现有机制词表无法表达该 proposal 的作用 | 需要 design review |
+| `llm_independent_discovery` | LLM 发现无法归入当前 harness abstraction 的新方向 | 只作为 observation，不直接进入 harness |
+
+## Review 原则
+
+1. 不接受 case-specific 数值表，例如“这个 case 试 `max-num-seqs=24`”。
+2. 若归类为 `same_operator_new_step`，只能修改通用 step policy，例如 grow/shrink factor、local grid budget、bracket 触发条件。
+3. 若归类为 `missing_descriptor`，descriptor 只能表达 knob 语义、约束、search geometry 和 directional effects，不能表达具体目标答案。
+4. 任何被接受的 gap 都需要新增 synthetic test，证明它不依赖 vLLM 常见取值或某个 bad-start case。
+
+## Pending
+
+当前 repo 内尚无已人工接受的 candidate family gap。实验产生的 JSON artifact 需要在这里补充 review 摘要后再进入代码设计。

docs/harness-ablation/knob-conditional-effects-20260705.md

@@ -0,0 +1,192 @@
+# Knob conditional effect 证据整理
+
+本文整理 2026-07-01 到 2026-07-03 在 `dash1` 上跑的 interaction screening 结果，用来支持一个具体论点：
+
+> Serving tuning knobs 不是彼此独立的。一个 knob 的收益方向和收益大小依赖当前 topology、admission/concurrency 和 scheduler context，因此不能假设“逐个 knob tune 到最好”一定可靠。
+
+## Presentation review: 应该怎么展示
+
+原来的 delta summary 能证明 `Delta_knob(context)` 不同，但它不够直观，因为它没有展示 tuning algorithm 会怎么失败。更适合作为 paper 主图的是：
+
+1. **主图：OAT path counterexample**  
+   在同一个 measured response surface 上画两条 one-knob-at-a-time 路径。读者能直接看到：同一个起点、不同单维 tuning 顺序，会停在不同点，而且其中一个是 coordinate-wise local optimum。
+2. **补充图：interaction residual**  
+   用 additive model residual 形式说明：如果 TP 和 MNS 是独立贡献，残差应接近 0；实际残差有结构性正负块。
+3. **补充图：delta/context summary**  
+   保留为形式化证据，但不作为主图，因为它不能直接展示 OAT 的路径依赖。
+
+因此本文推荐把 `knob-oat-counterexample-c1-qwen30b` 作为主文图，把 C3 crossed lines 和 residual/delta 放在 appendix 或机制分析图中。
+
+## 图 1：OAT path counterexample
+
+![OAT counterexample](figures/knob-oat-counterexample-c1-qwen30b.png)
+
+数据来源：
+
+- `interaction-mixed-qwen30b-tp-mns-surface-high1-dash1-d8899c5-20260701T095858Z`
+- `interaction-mixed-qwen30b-tp4-mns-nocap-qps20-dash1-d8899c5-20260701T161900Z`
+
+这张图直接展示为什么“逐个维度独立 tune”不可靠。我们从同一个起点 `TP=1, MNS=8` 出发：
+
+| Strategy | Path | Final req/s/GPU |
+|---|---|---:|
+| tune MNS first, then TP | `TP1,MNS8 -> TP1,MNS16 -> TP4,MNS16` | `2.44` |
+| tune TP first, then MNS | `TP1,MNS8 -> TP2,MNS8 -> TP2,MNS32` | `3.28` |
+
+`TP4,MNS16` 是一个 measured coordinate-wise local optimum：
+
+- 固定 `TP=4` 调 `MNS`：`MNS16/32/64` 都是 `2.44`，没有 strictly improving move；
+- 固定 `MNS=16` 调 `TP`：`TP4=2.44` 高于 `TP1=2.35` 和 `TP2=2.27`；
+- 但全局最好点 `TP2,MNS32=3.28` 比它高 `25.6%`。
+
+这比单纯说 “MNS 的 delta 依赖 TP” 更有力：它展示了一个实际 tuning path 如何被独立维度假设带到次优点。要从 `TP4,MNS16` 逃到 `TP2,MNS32`，tuner 必须允许非独立的 context-aware move，或者至少维护 frontier/plateau 上的反事实 anchor；单维 greedy OAT 不够。
+
+这里的结论不是“所有 workload 都有强 interaction”，而是更严格地说：
+
+1. 在真实 case 中确实存在明显 conditional effect；
+2. 这个现象足以否定 naive one-knob-at-a-time/OAT 作为通用 tuning strategy；
+3. harness 需要维护 mechanism-aware context，而不是把 knobs 当作独立维度。
+
+## Formal definition
+
+记某个 engine config 的 SLO-feasible objective 为：
+
+```text
+f(config) = max request_rate_per_gpu subject to pass_rate >= target
+```
+
+对 knob `x` 的一个 intervention `x_low -> x_high`，在 context `c` 下的效果定义为：
+
+```text
+Delta_x(c) = f(x_high, c) - f(x_low, c)
+```
+
+如果存在两个 context `c1, c2`，使得：
+
+```text
+Delta_x(c1) != Delta_x(c2)
+```
+
+则说明 knob `x` 存在 conditional effect。若符号也变化，比如一个 context 下提升、另一个 context 下降，则是更强的 interaction。
+
+## 图 2：C1 Qwen30B mixed workload surface
+
+![C1 Qwen30B surface](figures/knob-conditional-c1-qwen30b-surface.png)
+
+数据来源：
+
+- `interaction-mixed-qwen30b-tp-mns-surface-high1-dash1-d8899c5-20260701T095858Z`
+- `interaction-mixed-qwen30b-tp4-mns-nocap-qps20-dash1-d8899c5-20260701T161900Z`
+
+关键观察：
+
+| Context | `MNS=8 -> 32` 的 req/s/GPU 变化 |
+|---|---:|
+| `TP=1` | `2.10 -> 2.28`, `+8.7%` |
+| `TP=2` | `2.28 -> 3.28`, `+44.3%` |
+| `TP=4` | `1.28 -> 2.44`, `+90.3%` |
+
+这说明 `max-num-seqs` 的收益强烈依赖 `tensor-parallel-size`。同一个 `MNS` 调整在 `TP=1` 下只是小幅提升，在 `TP=2/4` 下变成决定性能上限的关键 knob。
+
+反过来看，`TP` 的收益也依赖 `MNS`：
+
+- 在 `MNS=8` 时，`TP=4` 是坏点，只有 `1.28 req/s/GPU`；
+- 在 `MNS=32` 时，`TP=2` 变成全局最优附近，达到 `3.28 req/s/GPU`。
+
+因此，如果 tuner 固定 `MNS=8` 去判断 topology，会错误低估 `TP=4`，也会无法看到 `TP=2 + MNS=32` 的最佳区域；如果固定 `TP=1` 去调 `MNS`，又会低估更高 TP 下 concurrency knob 的价值。这就是 OAT order sensitivity。
+
+## 图 3：C1 additive residual
+
+![C1 interaction residual](figures/knob-interaction-residual-c1-qwen30b.png)
+
+如果 `TP` 和 `MNS` 可以独立建模，一个简单 additive model：
+
+```text
+f(TP, MNS) ~= base + effect(TP) + effect(MNS)
+```
+
+应该留下接近 0 的 residual。实际 residual 最大达到约 `0.46 req/s/GPU`，而且呈现结构性模式：
+
+- `TP2,MNS32/64` 是正 residual，说明这个组合比独立效应相加更好；
+- `TP2,MNS16` 和 `TP4,MNS8` 是强负 residual，说明某些组合显著低于独立假设预测。
+
+这张图适合放在机制/appendix 中，用数学形式支持“不是独立 knob effect”。
+
+## 图 4：C3 Qwen235B decode workload
+
+![C3 Qwen235B decode lines](figures/knob-conditional-c3-qwen235b-decode-lines.png)
+
+数据来源：
+
+- `interaction-qwen235b-decode-c3-topo-mns-mbt-fixed-dash1-d8899c5-20260703T022514Z`
+
+完整 8 点结果：
+
+| Config | req/s/GPU | pass rate |
+|---|---:|---:|
+| `TP4 DP2 EP8 MNS64 MBT256` | `0.0535` | `1.0000` |
+| `TP4 DP2 EP8 MNS64 MBT384` | `0.0535` | `0.9922` |
+| `TP4 DP2 EP8 MNS128 MBT256` | `0.0590` | `0.9929` |
+| `TP4 DP2 EP8 MNS128 MBT384` | `0.0590` | `0.9929` |
+| `TP2 DP4 EP8 MNS64 MBT256` | `0.0590` | `0.9753` |
+| `TP2 DP4 EP8 MNS64 MBT384` | `0.0535` | `0.9961` |
+| `TP2 DP4 EP8 MNS128 MBT256` | `0.0590` | `0.9788` |
+| `TP2 DP4 EP8 MNS128 MBT384` | `0.0590` | `0.9823` |
+
+关键观察：
+
+- `MBT 256 -> 384` 在 `TP4/DP2 + MNS64` 下没有收益；
+- 同一个 `MBT 256 -> 384` 在 `TP2/DP4 + MNS64` 下反而下降约 `9.2%`；
+- `MNS 64 -> 128` 在 `TP4/DP2` 下提升约 `10.1%`；
+- 同一个 `MNS 64 -> 128` 在 `TP2/DP4 + MBT256` 下没有收益，但在 `TP2/DP4 + MBT384` 下恢复约 `10.1%`。
+
+这说明 runtime knobs 的作用不是单调独立的。`MBT` 是否有害取决于 topology 和 `MNS`；`MNS` 是否有用也取决于 topology 和 `MBT`。
+
+## 图 5：Delta 形式的直接证据
+
+![Delta summary](figures/knob-conditional-delta-summary.png)
+
+这张图把上面的论证直接转成 `Delta_x(context)`：
+
+- C1 中，同样是 `MNS 8 -> 32`，收益从 `+8.7%` 到 `+90.3%` 不等；
+- C3 中，同样是 `MBT 256 -> 384`，有的 context 是 `0%`，有的 context 是 `-9.2%`；
+- C3 中，同样是 `MNS 64 -> 128`，有的 context 是 `0%`，有的 context 是 `+10.1%`。
+
+这就是 conditional effect 的直接测量证据。
+
+## C2 是边界案例，不是反例
+
+C2 Qwen235B prefill tight SLO 的结果更弱：
+
+- `TP4` family: `0.1067~0.1175 req/s/GPU`；
+- `TP8` family: `0.1727 req/s/GPU`；
+- 在测过的 `MNS={64,128}`、`MBT={8192,16384}` 网格里 runtime knobs 基本平。
+
+这个 case 说明并不是每个 workload 都会在 runtime knobs 上表现出强 interaction。它的主要结论是 topology 主导：`TP8` 相比 `TP4` 约 `+47% req/s/GPU`。
+
+这对 paper framing 反而有用：我们的 claim 不应该是“所有 knobs 总是强耦合”，而应该是：
+
+> Tuning system 不能预设 knobs 独立；它必须通过 measured response 判断当前 case 是 topology-dominant、runtime-interaction-dominant，还是 flat/noisy。Harness 的作用是把这些 measured evidence 维护成 search context。
+
+## 对 harness 设计的含义
+
+这些图支持我们当前 framing：
+
+1. Harness 不应该只做单 knob local search。它需要保留 topology/runtime context，并允许 joint or projected interventions。
+2. Candidate generation 不能只说“把某个 knob 调大/调小”，而要说明这个 intervention 所依赖的 context。
+3. Validator 不能只比较 raw request rate；必须比较 SLO-feasible `request_rate_per_gpu`，并保存 negative evidence。
+4. LLM/planner 的价值不应被描述成“猜一个更好的 knob 值”，而是基于 harness 提供的 measured context 去提出 plausible joint moves。
+
+## 复现图
+
+```bash
+python3 scripts/plot_knob_conditional_effects.py
+```
+
+输出：
+
+- `docs/harness-ablation/figures/knob-oat-counterexample-c1-qwen30b.png`
+- `docs/harness-ablation/figures/knob-interaction-residual-c1-qwen30b.png`
+- `docs/harness-ablation/figures/knob-conditional-c1-qwen30b-surface.png`
+- `docs/harness-ablation/figures/knob-conditional-c3-qwen235b-decode-lines.png`
+- `docs/harness-ablation/figures/knob-conditional-delta-summary.png`

docs/harness-ablation/no-llm-harness-mechanism-20260625.md

@@ -0,0 +1,482 @@
+# No-LLM Harness Mechanism - 2026-06-25
+
+Status note, 2026-06-26:
+
+本文记录的是当前 rule-based prototype harness 的 no-LLM 机制和已有实验现象。它能证明
+AITuner 可以在没有 LLM endpoint 的情况下闭环运行，但不能证明 harness 的完备性、
+通用 robustness 或最终系统贡献。最终目标设计已经调整为 declarative intervention
+grammar + coverage-relative validator，见
+[`declarative-intervention-harness-design-20260626.md`](declarative-intervention-harness-design-20260626.md)。
+
+本文回答一个核心问题：如果不调用 LLM，harness 为什么还能自动找到配置？
+
+结论先说清楚：no-LLM 模式下并不是“没有 planner”。当前 harness 本身就是一个
+deterministic planner。LLM 在 AITuner 里只是一个可替换的 proposal backend；当
+harness 能从观测、瓶颈归因、候选 family 和 stop validator 中推出下一步时，tuning
+loop 会直接使用 harness proposal，而不会请求 LLM。
+
+## Tune loop 中 LLM 的位置
+
+`study tune` 每轮的决策顺序是：
+
+```text
+state + study spec + workload/probe results
+        |
+        v
+build_harness_context(...)
+        |
+        +--> build_harness_stop_proposal(context)
+        |       if true: write harness-stop and exit
+        |
+        +--> build_harness_guided_proposal(context)
+        |       if true: run this deterministic proposal
+        |
+        +--> call_llm_for_proposal(...)
+                only if no harness stop/proposal exists
+```
+
+因此在 `study.llm.endpoint = null` 的 no-LLM run 中，只要 harness 每轮都能给出
+一个 deterministic proposal 或 deterministic stop，整个实验就可以完全不调用 LLM。
+如果 harness 既不能 propose 也不能 stop，且没有 LLM endpoint，AITuner 会报错，而不是
+偷偷退化成随机搜索。
+
+当前 Qwen30B stopfix run 就是这种完整闭环：
+
+```text
+.aituner/qwen30b-harness-only-medium-stopfix-dash1-20260624T144701Z/
+```
+
+它没有 LLM endpoint，但仍完成了 9 个 measured trials，并最终由 validator 写出
+`harness_stop`。
+
+## Harness 做的不是 prompt engineering
+
+Harness 做的事情可以形式化成：
+
+```text
+H = (O, B, G, S, V)
+
+O: Observation schema
+   将 workload、trial probes、SLO failure、launch failure、topology constraints
+   转成结构化状态。
+
+B: Bottleneck attribution
+   将 SLO violation 归因到 serving regime，例如 ttft_prefill、decode_tpot、
+   admission_or_queueing、launch_or_memory。
+
+G: Intervention grammar
+   将 raw knobs 组织成有语义的 candidate families，例如 topology、batching、
+   sequence admission、KV memory headroom。
+
+S: Scoring policy
+   对候选 intervention 评分，选择最有信息量且最可能提升 SLO-constrained
+   req/s/GPU 的下一步。
+
+V: Validator / stop policy
+   阻止非法、重复、已知失败或无意义的 proposal；只有在剩余高价值候选被测完后
+   才允许 stop。
+```
+
+LLM 可以读取这些结构化信息并生成 proposal，但 no-LLM 时 `H` 自己就能生成
+proposal。换句话说，我们的核心是把：
+
+```text
+raw config vector search
+```
+
+转成：
+
+```text
+mechanism-guided intervention search
+```
+
+这就是为什么没有 LLM 也能工作。
+
+## Agent loop 流程图
+
+```mermaid
+flowchart TD
+  A[Baseline or latest measured trial] --> B[Load probe history and trial result]
+  B --> C[Build workload L-C-A profile]
+  B --> D[Build TrialProfile]
+  C --> E[Rank bottleneck hypotheses]
+  D --> E
+  E --> F[Generate legal candidate actions]
+  F --> G[Score candidates]
+  G --> H{High-value candidate?}
+  H -- yes --> I[Emit harness-proposal]
+  I --> J[Run real vLLM trial over search range]
+  J --> B
+  H -- no --> K{Validator stop allowed?}
+  K -- yes --> L[Emit harness-stop]
+  K -- no --> M{LLM endpoint exists?}
+  M -- yes --> N[Ask LLM backend]
+  M -- no --> O[Fail loudly: no proposal source]
+```
+
+## Observation: harness 看到什么
+
+每一轮 harness 不看自然语言日志做猜测，而是读结构化状态：
+
+- `StudySpec`
+  - hardware: GPU 数量、GPU 型号；
+  - engine: base flags/envs、tunable flags/envs、topology constraints；
+  - trace: request mode、window id、输入长度过滤、输出长度 override；
+  - SLO: TTFT/TPOT rule、target pass rate；
+  - search: load range、tolerance、probe budget。
+- `window_summary` / `WorkloadProfile`
+  - L: request length 分布、tail ratio；
+  - C: prefix/cache reuse；
+  - A: arrival rate、burstiness、interarrival variation。
+- 最近 trials
+  - config patch；
+  - best feasible request rate；
+  - request_rate_per_gpu；
+  - pass rate；
+  - probe history；
+  - latency p50/p95/p99；
+  - SLO failure reason counts；
+  - launch/runtime failure stage。
+
+这些数据会被压成 `recent_trial_diagnostics` 和 `trial_profiles`，后续步骤只使用这些结构化
+字段。
+
+## Bottleneck classifier: 怎么判断方向
+
+Harness 维护一组 ranked bottleneck hypotheses：
+
+```text
+ttft_prefill
+decode_tpot
+admission_or_queueing
+launch_or_memory
+```
+
+它的输入不是单一阈值，而是多类证据：
+
+- workload default：长 prompt tail 更偏向 `ttft_prefill`；
+- request mode：decode-only 且有 TPOT SLO 时更偏向 `decode_tpot`；
+- probe sequence：最近 trial 的 active bottleneck 权重大于旧 trial；
+- failed reason counts：
+  - `ttft_ms>...` 支持 `ttft_prefill`；
+  - `tpot_ms>...` 支持 `decode_tpot`；
+  - `arrival_lag_s>` / `probe_elapsed_s>` 支持 `admission_or_queueing`；
+- launch failure / OOM：支持 `launch_or_memory`。
+
+代码里这不是一个硬编码单标签，而是带 confidence 的 ranked list。例如最近 probe
+明确出现 TPOT failure，会提高 `decode_tpot` 分数；如果同时 workload 有长 prompt tail，
+`ttft_prefill` 仍会保留为次级 hypothesis。
+
+## Candidate family: raw knobs 如何变成 intervention
+
+Harness 不直接在所有 tunable flags 上盲采样。它先把 knobs 分成有系统含义的
+intervention family：
+
+| Family | 代表 knobs | 机制含义 |
+| --- | --- | --- |
+| topology | `tensor-parallel-size`, `data-parallel-size`, EP knobs | 改变每请求并行度、replica 数量、通信/效率 tradeoff |
+| batching | `max-num-batched-tokens`, `enable-chunked-prefill` | 改变 prefill/decode batching 与 HoL blocking |
+| admission | `max-num-seqs` | 改变并发 admission 与 TPOT/TTFT tail |
+| KV memory | `gpu-memory-utilization` | 改变 KV cache blocks 和可承载并发 |
+| failure memory | failed signatures | 阻止重复已知 launch/runtime 失败方向 |
+
+关键点是：candidate 来自当前 `StudySpec` 的 tunable schema 和 topology constraints。
+例如 topology candidate 只枚举合法 TP/DP/EP 组合；如果 EP 没有直接证据，generic
+topology search 不会主动引入 EP。
+
+## Scoring: 为什么会先走 topology，再走 gmu
+
+Candidate action 的评分大致是：
+
+```text
+score = expected_bottleneck_relief * bottleneck_confidence
+      + information_gain
+      + launch_safety
+      - regression_risk
+```
+
+然后 `experiment_plan.next_action` 选择最高分候选。分数超过阈值时，harness 直接生成
+proposal；否则进入 stop validator 或 LLM fallback。
+
+这套 scoring 体现了几个系统原则：
+
+1. Topology 是 serving 的一阶决策。
+   当 TP frontier 还没测完，`gpu-memory-utilization`、`max-num-seqs` 这类 runtime
+   微调不会抢在 topology 前面。
+
+2. Topology 不是“越大越好”。
+   评分和最终 winner 都看 `request_rate_per_gpu`，不是 raw request rate。TP4 可能总吞吐
+   更高，但如果使用更多 GPU 后 per-GPU 效率下降，就不会成为 incumbent。
+
+3. Runtime tuning 必须 anchored on incumbent topology。
+   当 topology 已经验证过，runtime proposal 会 preserve 当前 best topology，只在其上
+   调 `gpu-memory-utilization`、`max-num-seqs`、`max-num-batched-tokens`。
+
+4. Measurement 决定最终答案。
+   Candidate 只是一个 hypothesis；是否接受由真实 trial 的 SLO-constrained
+   `request_rate_per_gpu` 决定。
+
+5. Bad-start recovery 需要先 bracket，再微调。
+   如果 no-LLM run 从一个很高 TP 的初始点开始，且同 DP 下更高 TP frontier 已经不存在
+   或已测过，harness 会优先验证相邻低 TP，而不是把当前高 TP 当作 topology 已收敛。
+   这避免了 `TP=8` 这类坏初始点直接进入 `gpu-memory-utilization` 微调。
+
+6. Pathological runtime 起点需要跳回正常工作区间。
+   `gpu-memory-utilization` 的常规策略是在 settled topology 上小步 hill-climb；
+   但如果初始值明显低于正常工作区间，例如 `0.5`，harness 会先跳到 nominal floor
+   `0.9`，再按 `0.02` 步长向 safe ceiling `0.97` 验证。
+
+## Validator stop: 为什么不会过早停止
+
+Harness stop 不是“找到一个不错配置就停”。当前 stop validator 包含几个条件：
+
+- `search_high_saturated_by_incumbent`
+  - incumbent 的最高 feasible probe 已经贴近 configured search high；
+  - 说明当前测量范围已被打满。
+- `topology_frontier_requires_probe`
+  - 如果 active bottleneck 仍要求更高 TP frontier 且未测，禁止 stop。
+- `experiment_plan_has_high_value_candidate`
+  - 如果还有高分候选，禁止 stop。
+- `post_incumbent_validation_exhausted`
+  - strong incumbent 后至少要有 post-incumbent validation；
+  - validation 覆盖 topology/runtime family 或达到足够数量；
+  - 没有任何 validation trial 超过 incumbent；
+  - 才允许 clean stop。
+
+所以 validator 的作用是 fail-safe：
+
+```text
+wrong proposal 最多浪费一个 trial；
+wrong stop 会终止搜索，所以必须由 deterministic validator 授权。
+```
+
+## Qwen30B no-LLM run 中具体发生了什么
+
+Run:
+
+```text
+qwen30b-harness-only-medium-stopfix-dash1-20260624T144701Z
+```
+
+设置：
+
+- Model: `Qwen/Qwen3-30B-A3B`
+- Engine: community vLLM 0.20
+- Hardware: 8x H20, 允许 TP/DP/EP frontier
+- Trace: chat 0-8k, output 128, replay time scale 0.1
+- SLO: target pass rate 0.95, TTFT step rule, TPOT 50ms
+- LLM endpoint: `null`
+
+真实 trial path:
+
+| Trial | Source | Config patch | req/s/GPU | pass rate | Harness 解释 |
+| --- | --- | --- | ---: | ---: | --- |
+| 0001 | baseline | `{}` | 2.2000 | 1.0000 | 建立 baseline 和 probe evidence |
+| 0002 | harness | `TP=2` | 3.2583 | 1.0000 | latency/SLO pressure 下先测 adjacent TP |
+| 0003 | harness | `TP=4` | 2.0917 | 1.0000 | 验证更高 TP frontier；raw 总吞吐高但 per-GPU 低 |
+| 0004 | harness | `TP=2, gmu=0.92` | 3.2583 | 1.0000 | topology 已 settle，开始 incumbent topology 上的 KV headroom climb |
+| 0005 | harness | `TP=2, gmu=0.94` | 3.2583 | 1.0000 | 继续小步 hill-climb；未改善但未失败 |
+| 0006 | harness | `TP=2, gmu=0.96` | 3.3333 | 1.0000 | KV headroom 带来更高 feasible frontier |
+| 0007 | harness | `TP=2, gmu=0.97` | 3.4333 | 1.0000 | 达到 safe ceiling，成为 incumbent |
+| 0008 | harness | `TP=4, DP=2` | 1.0458 | 1.0000 | post-incumbent topology validation，没有超过 incumbent |
+| 0009 | harness | `TP=8` | 1.0458 | 1.0000 | 继续 frontier validation，没有超过 incumbent |
+| 0010 | harness stop | stop | - | - | validator: `post_incumbent_validation_exhausted` |
+
+这个过程里没有外部 LLM 决策。每一步 proposal 都来自 harness：
+
+1. baseline 观测到当前 engine 在 SLO 下的可行 frontier；
+2. bottleneck/机制模型认为 topology 是一阶干预；
+3. 测 TP2，接受，因为 per-GPU 从 2.2 提到 3.2583；
+4. 测 TP4，拒绝为 incumbent，因为 per-GPU 降到 2.0917；
+5. topology frontier settle 后，在 TP2 上小步提升 `gpu-memory-utilization`；
+6. `gmu=0.97` 达到 3.4333；
+7. 再测 nearby topology，确认没有更好；
+8. validator 授权 stop。
+
+## 为什么这不是写死 Qwen30B
+
+这条路径看起来像“harness 知道答案是 TP2+gmu0.97”，但代码机制不是这样写的。
+
+没有写死的部分：
+
+- 没有写死 model name；
+- 没有写死 Qwen30B；
+- 没有写死 `TP=2` 是最终答案；
+- 没有写死 `gmu=0.97` 一定最好；
+- 没有跳过真实测量；
+- 没有把 TP4/TP8 直接判负，而是实际运行并比较。
+
+真正写入 harness 的 domain knowledge 是：
+
+- TP/DP/EP 是 topology family，必须满足 topology constraints；
+- topology 通常是一阶 serving intervention，要先于 runtime 微调被验证；
+- raw throughput 不等于目标，跨 topology 比较要用 `request_rate_per_gpu`；
+- `gpu-memory-utilization` 是 KV memory headroom 微调，只应在 incumbent topology 上小步 hill-climb；
+- launch failure 和 tested signatures 是 hard negative evidence；
+- stop 必须由 validator 授权，不能由 proposer 自己说停就停。
+
+这是一种系统机制约束，不是 case-specific prompt。
+
+## 它和 BO / raw heuristic 的区别
+
+普通 BO 或 raw heuristic 的搜索空间通常是：
+
+```text
+config = {tp, dp, ep, gmu, max_num_seqs, max_num_batched_tokens, ...}
+score = measured req/s/GPU
+```
+
+这会产生几个问题：
+
+- 它不知道哪些 knobs 是 topology family，哪些是 runtime family；
+- 它可能在没测 TP frontier 前浪费大量 trial 调 runtime；
+- 它可能重复已知 launch failure；
+- 它可能把 raw throughput 高但 GPU efficiency 差的配置误当方向；
+- 它很难解释“这个 trial 试图证伪哪个瓶颈 hypothesis”。
+
+Harness-shaped search space 是：
+
+```text
+state -> bottleneck hypothesis -> legal intervention family -> measured verdict
+```
+
+因此 BO、bandit、LLM、deterministic heuristic 都可以接在 harness 后面。它们优化的不是
+raw knob vector，而是有 serving 语义的 intervention graph。
+
+这也是我们新 framing 的核心：
+
+```text
+black-box optimization
+  -> grey-box / mechanism-guided experimental optimization
+```
+
+## 当前还需要补的证据
+
+No-LLM Qwen30B run 证明了 deterministic harness 可以完整闭环，但 paper 还需要继续补：
+
+1. Planner-agnostic ablation
+   - `raw BO` vs `harness-guided BO`；
+   - `raw heuristic` vs `harness deterministic policy`；
+   - 证明收益来自 harness substrate，而不是某个 LLM。
+
+2. Mechanism ablation
+   - no attribution；
+   - shuffled attribution；
+   - no topology-first；
+   - no intervention grammar；
+   - no validator/failure memory。
+
+3. Near-optimum evidence
+   - 在 1-2 个 case 做局部 grid；
+   - 证明 harness path 找到的是 near-optimal region，不只是一个可行 config。
+
+4. Cross-case robustness
+   - 再选 decode-heavy 或 long-prefill case；
+   - 验证不同 workload/SLO 下 candidate family 会发生合理切换。
+
+5. Bad-start recovery
+   - 从非可信初始配置开始，例如 `TP=8, max-num-seqs=8, gmu=0.5`；
+   - 证明 harness 不是只能从“已经比较合理”的 base config 出发；
+   - 观察它是否能先恢复 topology，再恢复 runtime headroom，并最终回到同一 near-optimal
+     region。
+
+## Bad-start recovery 审计 - 2026-06-26
+
+用户提出的问题是：如果我们不是从可信 base config 开始，而是从一个恶意或不合理的
+配置开始，例如：
+
+```text
+TP=8, DP=1, max-num-seqs=8, gpu-memory-utilization=0.5
+```
+
+no-LLM harness 是否仍能自动找到正确方向？
+
+目前结论要分开说：
+
+1. **旧 planner 不能直接 claim 任意坏起点可恢复。**
+   本地合成审计显示，旧逻辑会把 `TP=8` 误当作 topology frontier 已收敛，并把下一步
+   proposal 设为 `gpu-memory-utilization=0.52`。这会在坏 topology 和坏 runtime 上
+   做很慢的小步爬坡，不能作为 robust evidence。
+
+2. **已补 planner 机制。**
+   当前 harness 增加了两个 no-LLM deterministic recovery rules：
+   - `bad_start_topology_bracket`：当当前 anchor 在高 TP，且没有未测的更高 TP frontier 时，
+     先测相邻低 TP，例如 `TP=8 -> TP=4`；
+   - `gmu_nominal_floor`：当 settled topology 上的 `gpu-memory-utilization < 0.9` 时，
+     先跳到 `0.9`，再做常规 `0.92/0.94/.../0.97` hill-climb。
+
+3. **已加本地回归测试，但还没做真机证明。**
+   已通过的 planner tests：
+   - `test_harness_brackets_down_from_bad_high_tp_start_before_runtime_tuning`
+   - `test_harness_jumps_low_gpu_mem_util_to_nominal_floor_after_topology_settles`
+   - 以及已有 topology-first / gmu-climb 相关回归测试。
+
+因此，当前状态是：planner 侧已经能给出正确方向；paper 级别还需要真机 bad-start
+recovery run 来确认真实 vLLM 测量下是否稳定收敛。
+
+## 准备中的真机实验
+
+实验目的不是再证明默认起点能 work，而是证明：
+
+```text
+same workload + same SLO + same no-LLM harness
+不同初始 config
+  -> 是否收敛到同一 near-optimal region
+  -> 是否保持可解释 trial path
+```
+
+Base spec 使用已验证的 Qwen30B community vLLM 0.20 harness setup：
+
+```text
+configs/examples/dash0_qwen30b_a3b_community_vllm020_harness.json
+```
+
+运行时需要设置：
+
+```json
+{
+  "llm": {
+    "use_harness": true,
+    "endpoint": null
+  }
+}
+```
+
+建议最小矩阵：
+
+| Case | Base flags 变化 | 要验证的机制 | 预期 trial path |
+| --- | --- | --- | --- |
+| trusted-start-control | 保持现有可信 base | 对照已有 stopfix run | `TP=2 -> TP=4 -> TP=2+gmu climb -> stop` |
+| bad-topology | `TP=8, DP=1` | 高 TP 起点是否会向下 bracket | `TP8 baseline -> TP4 -> TP2/或同等 better topology -> runtime` |
+| bad-runtime | `TP=2, DP=1, gmu=0.5, max-num-seqs=8` | 低 KV headroom 是否跳回正常区间 | `gmu 0.5 baseline -> gmu 0.9 -> 0.92/...` |
+| combined-bad | `TP=8, DP=1, gmu=0.5, max-num-seqs=8` | topology recovery 和 runtime recovery 能否串起来 | `TP8 -> TP4 -> TP2/nearby -> gmu 0.9 -> climb -> stop` |
+
+成功判据：
+
+- 不配置 LLM endpoint；所有 proposal 来自 harness；
+- 不重复相同 config signature；
+- high-TP 起点必须先出现相邻低 TP probe，而不是先做 `gmu=0.52`；
+- low-gmu 起点必须先跳到 `0.9`，而不是 `0.52`；
+- 在 12 个 measured trials 内达到 reference stopfix best 的 `>=95%`：
+
+```text
+reference best = 3.4333 req/s/GPU
+95% threshold  = 3.2616 req/s/GPU
+```
+
+- 最终 stop 必须是 validator 授权，例如 `harness_stop`，而不是因为没有 proposal source
+  失败退出。
+
+如果真机结果失败，需要保留失败路径并分析是哪类机制不足：
+
+- topology bracket 找到低 TP，但 runtime 仍无法恢复；
+- `max-num-seqs=8` 导致 admission 太差，需要 admission recovery floor；
+- baseline 自身全不可行，当前 harness 缺少 completed incumbent，不能进入正常 guided loop；
+- vLLM launch/OOM 造成 failure memory 覆盖了可恢复路径。
+
+## 一句话总结
+
+No-LLM harness 能自动找到配置，是因为它已经实现了一个面向 serving 机制的实验 planner：
+先把 trial 观测归因成 bottleneck，再把 bottleneck 映射成合法 intervention family，按
+SLO-constrained req/s/GPU 真实测量更新 incumbent，最后由 validator 判断是否停止。
+LLM 只是这个 planner 的一个可替换 proposal backend，而不是 AITuner 的必要核心。

docs/harness-ablation/prefill-scheduler-normalized-design-20260629.md

@@ -0,0 +1,336 @@
+# Prefill Scheduler Interaction Harness 设计与 Review
+
+日期：2026-06-29
+
+## 背景
+
+case3 的 ablation 结果显示，`gpt-5.5 no-harness` 找到了一个 runtime/scheduler 方向：
+
+```text
+enable-chunked-prefill=true
+max-num-batched-tokens 较低/中等
+max-num-seqs 适中
+block-size=16
+```
+
+而当时 harness 主要做两类动作：
+
+- 单点打开 `enable-chunked-prefill`；
+- 对 `max-num-batched-tokens` 做单调 raise。
+
+这个 gap 不能用“把 8192/32 这组值加入 candidate grid”来修补。那会把 case3 的答案硬编码成更大的候选表，仍然是 rule-based overfitting。
+
+## 设计原则
+
+新增的设计不是一个 fixed value set，而是一个 normalized control dimension：
+
+```text
+prefill_quantum_ratio = max-num-batched-tokens / prompt_tokens_p95
+admission_pressure    = max-num-seqs relative to trace.max_concurrency
+scheduler_mode        = enable-chunked-prefill
+```
+
+因此，candidate generator 不直接说“试 8192”，而是说：
+
+- 如果 long-tail prefill + TTFT/prefill bottleneck，且当前 `prefill_quantum_ratio` 太大，则沿 log-ratio 方向降低 prefill quantum；
+- 如果 prefill quantum 远小于 prompt scale，可能是过度切碎导致 overhead，则沿 log-ratio 方向提高 prefill quantum；
+- 如果 admission/queueing 是瓶颈，则只按 relative step 调整 admission pressure；
+- 所有 concrete flag value 都是最后一步从 normalized target 映射到 engine flag，并按 engine granularity round。
+
+当前实现使用几何中点作为 trust-region step：
+
+```text
+target_mbt = sqrt(current_mbt * prompt_tokens_p95)
+```
+
+这对应在 log space 走半步。它比固定乘以 0.5/1.5 更接近 scale-invariant：prompt scale 变大时，下一步 MBT 也会变大。
+
+## Agent Loop
+
+当前 harness 的 loop 可以形式化为：
+
+```text
+trial result
+  -> observation extractor
+  -> bottleneck classifier
+  -> candidate family selector
+  -> normalized candidate generator
+  -> scoring / coverage ranking
+  -> validator / no-repeat / stop guard
+  -> next trial
+```
+
+每一层承担不同责任：
+
+1. observation extractor 只把 trial result 转成可比较的事实，例如
+   request_rate_per_gpu、pass_rate、失败原因、TTFT/TPOT 分位数。
+2. bottleneck classifier 把事实归入 `ttft_prefill`、`decode_tpot`、
+   `admission_or_queueing` 等机制瓶颈，不直接输出配置值。
+3. candidate family selector 决定要验证哪个系统假设，例如 topology frontier、
+   prefill scheduler、admission pressure 或 GPU memory headroom。
+4. normalized candidate generator 才把机制变量映射成具体 engine flag。
+5. scoring / coverage ranking 负责排序：未覆盖但机制上相关的维度应优先于
+   已知方向上的微调。
+6. validator 使用 normalized full-config signature 防止重复测试，并用 stop guard
+   避免在仍有高价值 falsification candidate 时过早停止。
+
+因此，harness 的核心不是“把 LLM prompt 写好”，而是把黑盒搜索拆成带因果方向的
+white-box falsification loop。LLM 可以参与生成候选或解释候选，但候选必须通过
+harness 的 family、signature、scoring 和 validator 约束。
+
+## 实现映射
+
+代码入口：
+
+- `src/aituner/harness.py::_runtime_candidate_actions`
+  - 在 topology frontier settled 后调用新的 `_prefill_scheduler_candidate_actions`。
+  - 仍保留 topology-before-runtime guard，runtime family 不抢未覆盖的 topology frontier。
+
+新增逻辑：
+
+- `_prefill_scheduler_workload_applies`
+  - 只在非 decode-only、long-tail/moderate-tail prefill workload、非 high-prefix-reuse 场景激活。
+- `_next_prefill_quantum_step`
+  - 使用 `current_mbt / prompt_scale` 判断方向。
+  - 通过几何中点做相对 step。
+- `_next_admission_pressure_step`
+  - 使用 `max-num-seqs / trace.max_concurrency` 作为 normalized admission pressure。
+  - 当 admission/queueing 受限且 admission pressure 过低时 raise。
+  - 当 TTFT/prefill 受限且 admission pressure 明显高于 trace concurrency scale 时 lower。
+- `_prefill_scheduler_candidate_actions`
+  - 输出 `prefill-scheduler-interaction` family。
+  - `score_factors` 显式记录 current/target `prefill_quantum_ratio`，方便后续实验解释。
+  - `score_factors` 同时记录 current/target admission pressure ratio，避免只解释 MBT。
+  - 当 scheduler dimension 还没有被 materialized config 覆盖时，加入
+    `uncovered_scheduler_dimension_bonus`，让该 family 在 topology settled 后优先于
+    `gpu-memory-utilization` 这类 resource micro-tuning。
+- 当该 family 已生成有效候选时，旧的 standalone `raise_mbt`、
+  `enable_chunked_prefill`、`raise_mbt_and_max_num_seqs` 只作为 fallback，不作为同级
+  prefill runtime 候选抢排序。
+- `gpu-memory-utilization` 仍保留小步 hill-climb，但继续爬升必须由同拓扑
+  request_rate_per_gpu 改善支撑；仅仅 launch 成功或打平 incumbent 不再算成功。
+
+## 为什么不是 rule-based hack
+
+禁止的实现形态：
+
+- 不允许引用 case3、具体 trace 名、模型名、机器名；
+- 不允许出现 `if TP=2 and gmu=0.7 and mns=8 then MBT=8192`；
+- 不允许把 case3 发现扩成 `{4096,8192,12288,16384} x {16,32,64}` 这种固定 grid；
+- 不允许 bypass normalized full-config signature。
+
+当前实现满足：
+
+- trigger 来自 L-C-A profile、bottleneck classifier、topology frontier、tunable flags；
+- proposal 是相对当前 incumbent 的 direction，不是固定答案；
+- concrete value 随 prompt scale 和 current config 改变；
+- validator/no-repeat 仍使用 normalized effective full-config signature；
+- runtime gate 和正式 topology frontier 共用 higher-TP frontier patch 构造，避免
+  DP 非 base 时 scheduler 抢跑；
+- short prompt、decode-only、high prefix reuse 不激活该 family。
+
+但这不是完备性证明。当前能 claim 的是更严格的工程性质：
+
+- 不引用特定 case identity；
+- 不把已知 winner 写成候选表；
+- 每个 concrete proposal 都能追溯到一个 normalized mechanism variable；
+- 每次 trial 都能被解释成对一个系统假设的 falsification；
+- 失败时会留下可审计的 candidate sequence 和 score factors。
+
+## Review 结论
+
+### 之前实现的问题
+
+1. `enable-chunked-prefill` 是 standalone toggle，无法表达 scheduler interaction。
+2. TTFT/prefill bottleneck 下 MBT 主要单调 raise，无法发现“降低 prefill quantum 减少 HoL blocking”。
+3. 旧测试断言了固定 `16384` 等值，容易把 harness 叙事拉回 heuristic table。
+
+### 当前改动的效果
+
+1. 引入 `prefill-scheduler-interaction` 作为新的 mechanistic family。
+2. candidate 的 action id 表达方向：
+   - `lower_prefill_quantum_with_chunked_prefill`
+   - `raise_prefill_quantum_with_chunked_prefill`
+   - `seed_chunked_prefill_quantum`
+   - `adjust_admission_pressure_with_chunked_prefill`
+3. 测试改为验证 normalized direction 和 scale sensitivity，而不是固定 absolute value。
+
+### 当前实现仍需警惕的风险
+
+1. `_PREFILL_QUANTUM_HEAD_OF_LINE_RATIO=1.0` 和
+   `_PREFILL_QUANTUM_FRAGMENTATION_RATIO=0.5` 仍是机制阈值，不是定理。
+   它们必须通过 scaled prompt / negative workload 实验验证，而不能只靠 case3。
+2. `uncovered_scheduler_dimension_bonus` 是 coverage 排序策略。它的合理性来自
+   “先覆盖未 materialized 的机制维度，再做 GMU 微调”，但必须通过 candidate
+   sequence 证明它不会在 topology frontier 未覆盖时抢跑。
+3. `block-size=16` 目前没有被纳入这个 family。不能把它作为 case3 固定答案加入；
+   如果后续要处理，需要单独设计 allocator/layout family，从 engine capability 和
+   memory block waste observation 推导，而不是在 prefill scheduler family 里硬塞。
+4. 现有实现仍保留旧的 standalone `enable-chunked-prefill` 和 `raise_mbt` 路径作为
+   fallback。它们不能在 `prefill-scheduler-interaction` 已生成有效候选时作为同级
+   prefill runtime 候选抢排序。
+
+### 2026-06-29 独立 review 后的修正
+
+独立 review 指出了三个需要立即收紧的泛化风险：
+
+1. 旧 standalone MBT/chunked 候选仍可能让整体 harness 表现得像 heuristic table。
+2. admission pressure 只有 raise，没有处理 `max-num-seqs` 过高导致 TTFT/prefill 干扰。
+3. runtime gate 的 topology-settled 判断和正式 topology frontier 在 DP 非 base 时不完全一致。
+
+对应修正：
+
+- 当 `prefill-scheduler-interaction` 有有效候选时，旧的 standalone MBT/chunked/joint
+  prefill-runtime 候选降为 fallback。
+- admission pressure 改为 normalized ratio，并支持 raise/lower 两个方向：
+  `raise_admission_pressure_with_chunked_prefill` 和
+  `lower_admission_pressure_with_chunked_prefill`。
+- 抽出 `_higher_tp_frontier_patch`，让 runtime gate 与
+  `_topology_frontier_status` 使用同一套 higher-TP signature。
+- GMU hill-climb 改为 measurement-gated：同拓扑 GMU trial 没有提升
+  request_rate_per_gpu 时，阻断继续向更高 GMU 爬升，避免连续浪费 trials。
+
+### 2026-06-29 远端 review feedback
+
+在 dash1 用 `36c301c` 启动 case3 bad-runtime 重跑后，trial-0003 的
+candidate-set 已经出现 `prefill-scheduler-interaction`：
+
+```text
+action_id = seed_chunked_prefill_quantum
+patch     = enable-chunked-prefill=true, max-num-batched-tokens=8192
+ratio     = target prefill_quantum_ratio ~= 1.05
+```
+
+但初始 scoring 仍让 `raise_gpu_memory_utilization` 排在前面。这说明 family
+接入是正确的，但排序仍偏向 resource micro-tuning。随后实现加入
+`uncovered_scheduler_dimension_bonus`：当 topology frontier 已覆盖、prefill scheduler
+dimension 还没有被 materialized config 测过时，优先测试 scheduler hypothesis，
+避免重复旧 harness 先爬 GMU 的失败轨迹。
+
+## 单元验证
+
+新增/更新的测试覆盖：
+
+- long-tail TTFT 下，过大的 `prefill_quantum_ratio` 会下降；
+- prompt length scale 变大时，下一步 MBT target 也变大；
+- topology frontier 已覆盖后，未覆盖的 scheduler dimension 排在 GMU 微调之前；
+- short prompt workload 不激活 prefill scheduler family；
+- 原有 prefill stop guard 仍不允许在有 high-value candidate 时停止；
+- normalized full-config no-repeat 语义不变。
+
+本地全量测试：
+
+```text
+PYTHONPATH=src python3 -m unittest discover -s tests
+156 tests OK
+```
+
+本地重点回归：
+
+```text
+PYTHONPATH=src python3 -m unittest \
+  tests.test_core_flow.CoreFlowTests.test_prefill_scheduler_coverage_precedes_gmu_microtune \
+  tests.test_core_flow.CoreFlowTests.test_prefill_scheduler_admission_pressure_only_uses_normalized_seq_cap \
+  tests.test_core_flow.CoreFlowTests.test_prefill_scheduler_lowers_excess_admission_pressure \
+  tests.test_core_flow.CoreFlowTests.test_prefill_scheduler_negative_applicability_matrix \
+  tests.test_core_flow.CoreFlowTests.test_prefill_scheduler_does_not_preempt_open_topology_frontier \
+  tests.test_core_flow.CoreFlowTests.test_prefill_scheduler_lowers_quantum_by_normalized_ratio \
+  tests.test_core_flow.CoreFlowTests.test_prefill_scheduler_quantum_step_scales_with_prompt_length \
+  tests.test_core_flow.CoreFlowTests.test_prefill_scheduler_not_active_for_short_prompt_workload
+8 tests OK
+```
+
+## 还需要真机实验验证
+
+下一步实验不应该只看 case3 是否复现，而要攻击这个 family 的边界：
+
+1. case3 bad runtime start：
+   - 目标：验证 LLM+harness / no-LLM harness 是否能从 bad runtime start 找到 chunked-prefill scheduler 方向。
+2. scaled prompt case：
+   - 目标：验证 proposal 不固定在同一个 MBT，而会随 `prompt_tokens_p95` 改变。
+3. short/decode negative case：
+   - 目标：验证该 family 不会在不适用 workload 上误触发。
+4. topology frontier case：
+   - 目标：验证 topology 未覆盖时 runtime scheduler 不抢跑。
+
+核心指标：
+
+- best request_rate_per_gpu；
+- time-to-best / trial-to-target；
+- candidate family sequence；
+- `prefill_quantum_ratio_current -> target` 的方向是否与 bottleneck evidence 一致；
+- 是否出现 repeated normalized full-config signature。
+
+## 当前 dash1 真机状态
+
+当前正在验证提交 `bfd8579`：
+
+```text
+run     = .aituner/badstart-prefill-scheduler-bfd8579-20260628T173102Z
+case    = badstart-expanded-9accf25-20260626T184911Z-runtime_tp2_dp1_gmu070_mns8
+session = aituner-prefill-scheduler-case3-bfd8579
+```
+
+截至 2026-06-29 01:53 UTC+8 左右：
+
+- baseline trial-0001 已完成，best request_rate_per_gpu 约为 2.025；
+- trial-0002 TP4 topology frontier probe 已完成，best request_rate_per_gpu 约为 2.000，
+  没有超过 baseline；
+- candidate-set-0002 的 top action 是 topology frontier，符合 topology-before-runtime；
+- candidate-set-0003 的 top action 已变为 `seed_chunked_prefill_quantum`：
+
+```text
+score    = 0.69
+patch    = enable-chunked-prefill=true, max-num-batched-tokens=8192
+ratio    = prefill_quantum_ratio_target ~= 1.0536
+baseline = raise_gpu_memory_utilization score 0.64
+```
+
+这说明 `uncovered_scheduler_dimension_bonus` 达到了设计目的：topology frontier 覆盖后，
+未 materialized 的 scheduler dimension 会先于 GMU 微调被验证。
+
+trial-0003 已完成，best request_rate_per_gpu 约为 2.025，和 baseline 持平，没有形成
+性能提升。这个结果不能 claim scheduler seed 是 winner，但它提供了有价值的
+falsification evidence：coverage priority 改变了探索顺序，具体 `chunked + MBT ~= p95`
+hypothesis 被验证后没有改进。系统随后进入 candidate-set-0004，开始测试
+`gpu-memory-utilization=0.9`。trial-0004 同样完成在约 2.025，没有超过 baseline；
+trial-0005 的 `gpu-memory-utilization=0.92` 仍然打平 baseline，旧 run 随后继续排
+`gpu-memory-utilization=0.94`。这暴露出旧实现的 GMU hill-climb 问题：它把 launch
+成功当成 climb 成功，而没有要求 request_rate_per_gpu 改善。最新本地实现已经修正为
+measurement-gated GMU climb；下一轮应使用新提交重新跑，验证 GMU tie 后是否转向
+admission pressure、topology/DP 或其他 family。
+
+## Hardened Run Feedback
+
+使用提交 `6b25d56` 在 dash1 重新启动：
+
+```text
+run     = .aituner/badstart-prefill-hardened-6b25d56-20260628T180104Z
+case    = badstart-expanded-9accf25-20260626T184911Z-runtime_tp2_dp1_gmu070_mns8
+session = aituner-prefill-hardened-6b25d56
+```
+
+截至 2026-06-29 02:27 UTC+8 左右，同一 run 内的 trial sequence 是：
+
+| trial | patch | request_rate_per_gpu | observation |
+| --- | --- | ---: | --- |
+| 0001 | baseline bad-start | 2.983 | 同 run incumbent，明显高于旧 run baseline，说明跨 run 数字不能直接混用 |
+| 0002 | `tensor-parallel-size=4` | 1.629 | topology TP4 被 falsify |
+| 0003 | `enable-chunked-prefill=true, max-num-batched-tokens=8192` | 2.025 | standalone scheduler seed 被 falsify |
+| 0004 | `gpu-memory-utilization=0.9` | 3.258 | GMU=0.9 是当前 best，达到已知 no-harness 水平 |
+| 0005 | GMU=0.9 + scheduler seed | 2.025 | GMU 与 scheduler seed 的组合被 falsify |
+| 0006 | `gpu-memory-utilization=0.92` | 3.258 | 与 GMU=0.9 打平，没有继续提升 |
+| 0007 | `tensor-parallel-size=4, data-parallel-size=2` | 1.000 | DP/topology probe 被 falsify |
+
+candidate-set-0007 没有继续提出 `gpu-memory-utilization=0.94`，而是转向
+`tensor-parallel-size=4, data-parallel-size=2` topology probe。这验证了
+measurement-gated GMU climb：GMU=0.92 只是打平时，不再继续向更高 GMU 盲目爬升。
+candidate-set-0008 在 TP4/DP2 被 falsify 后继续测试 `tensor-parallel-size=8`。
+
+当前最重要的机制结论：
+
+- scheduler seed 的 priority 和 no-repeat 都按设计工作；
+- scheduler seed 在这个 case 不是独立 winner，必须被 measurement falsify；
+- GMU=0.9 是当前真正有效的机制维度；
+- GMU 的后续 climb 已经从 launch-gated 修正为 improvement-gated；
+- 后续应看 topology/DP、MNS 或 allocator/layout family 是否能进一步超过 3.258。

scripts/plot_knob_conditional_effects.py

@@ -0,0 +1,381 @@
+#!/usr/bin/env python3
+"""Plot measured knob conditional effects for the AITuner harness study."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import matplotlib
+
+matplotlib.use("Agg")
+import matplotlib.pyplot as plt
+import numpy as np
+from matplotlib.lines import Line2D
+from matplotlib.patches import Patch
+
+
+OUT = Path("docs/harness-ablation/figures")
+
+
+def save(fig: plt.Figure, name: str) -> None:
+    OUT.mkdir(parents=True, exist_ok=True)
+    fig.savefig(OUT / f"{name}.png", dpi=220, bbox_inches="tight")
+    fig.savefig(OUT / f"{name}.svg", bbox_inches="tight")
+
+
+def plot_c1_surface() -> None:
+    # Qwen30B mixed workload, TP x MNS screen. Values are req/s/GPU.
+    # Source runs:
+    # - interaction-mixed-qwen30b-tp-mns-surface-high1-dash1-d8899c5-20260701T095858Z
+    # - interaction-mixed-qwen30b-tp4-mns-nocap-qps20-dash1-d8899c5-20260701T161900Z
+    mns = np.array([8, 16, 32, 64])
+    tp = np.array([1, 2, 4])
+    values = np.array(
+        [
+            [2.1000, 2.3500, 2.2833, 2.2833],
+            [2.2750, 2.2750, 3.2833, 3.2583],
+            [1.2833, 2.4417, 2.4417, 2.4417],
+        ]
+    )
+
+    fig, axes = plt.subplots(1, 2, figsize=(12.8, 4.8), gridspec_kw={"width_ratios": [1.05, 1.2]})
+    ax = axes[0]
+    im = ax.imshow(values, cmap="YlGnBu", aspect="auto", vmin=1.2, vmax=3.35)
+    ax.set_xticks(range(len(mns)), labels=mns)
+    ax.set_yticks(range(len(tp)), labels=[f"TP={x}" for x in tp])
+    ax.set_xlabel("max-num-seqs (MNS)")
+    ax.set_ylabel("tensor-parallel-size")
+    ax.set_title("C1 response surface: req/s/GPU")
+    for i in range(values.shape[0]):
+        for j in range(values.shape[1]):
+            color = "white" if values[i, j] > 2.75 else "black"
+            ax.text(j, i, f"{values[i, j]:.2f}", ha="center", va="center", color=color, fontsize=10)
+    cbar = fig.colorbar(im, ax=ax, fraction=0.046, pad=0.04)
+    cbar.set_label("req/s/GPU")
+
+    ax = axes[1]
+    colors = {1: "#4E79A7", 2: "#59A14F", 4: "#E15759"}
+    for idx, t in enumerate(tp):
+        ax.plot(mns, values[idx], marker="o", linewidth=2.4, color=colors[int(t)], label=f"TP={t}")
+    ax.set_xscale("log", base=2)
+    ax.set_xticks(mns, labels=mns)
+    ax.set_xlabel("max-num-seqs (MNS)")
+    ax.set_ylabel("req/s/GPU")
+    ax.set_title("Non-parallel lines imply interaction")
+    ax.grid(True, axis="y", alpha=0.28)
+    ax.legend(frameon=False)
+    ax.annotate(
+        "TP=2 only becomes best\nwhen MNS reaches 32",
+        xy=(32, 3.2833),
+        xytext=(20, 3.05),
+        arrowprops={"arrowstyle": "->", "lw": 1.2},
+        fontsize=9,
+    )
+    ax.annotate(
+        "TP=4 is bad at MNS=8\nbut recovers at MNS>=16",
+        xy=(8, 1.2833),
+        xytext=(10, 1.55),
+        arrowprops={"arrowstyle": "->", "lw": 1.2},
+        fontsize=9,
+    )
+    fig.suptitle("Knob effects are conditional: MNS effect depends on TP", fontsize=14, y=1.02)
+    fig.tight_layout()
+    save(fig, "knob-conditional-c1-qwen30b-surface")
+    plt.close(fig)
+
+
+def plot_c1_oat_counterexample() -> None:
+    # C1 Qwen30B: one-knob-at-a-time tuning gets trapped at a coordinate-wise
+    # local optimum 25.6% below the measured global best. The right panel zooms
+    # into the trap's neighbourhood so the reader can SEE that every single-knob
+    # move from the trap is worse or flat, instead of having to read a caption.
+    mns = [8, 16, 32, 64]
+    tp = [1, 2, 4]
+    values = np.array(
+        [
+            [2.1000, 2.3500, 2.2833, 2.2833],
+            [2.2750, 2.2750, 3.2833, 3.2583],
+            [1.2833, 2.4417, 2.4417, 2.4417],
+        ]
+    )
+    idx = {(t, s): (mns.index(s), tp.index(t)) for t in tp for s in mns}
+
+    fig, axes = plt.subplots(1, 2, figsize=(14.0, 6.2), gridspec_kw={"width_ratios": [1.5, 1.0]})
+    ax = axes[0]
+    im = ax.imshow(values, cmap="YlGnBu", aspect="auto", vmin=1.2, vmax=3.35)
+    ax.set_xticks(range(len(mns)), labels=mns)
+    ax.set_yticks(range(len(tp)), labels=[f"TP={x}" for x in tp])
+    ax.set_xlabel("max-num-seqs (MNS)", fontsize=11)
+    ax.set_ylabel("tensor-parallel-size", fontsize=11)
+    ax.set_title("Two OAT paths from the same start", fontsize=12, loc="left")
+    for i in range(values.shape[0]):
+        for j in range(values.shape[1]):
+            color = "white" if values[i, j] > 2.75 else "black"
+            ax.text(j, i, f"{values[i, j]:.2f}", ha="center", va="center", color=color, fontsize=11, weight="bold")
+
+    def draw_path(path: list[tuple[int, int]], color: str, labels: list[str]) -> None:
+        for (a, b) in zip(path, path[1:]):
+            x0, y0 = idx[a]
+            x1, y1 = idx[b]
+            ax.annotate(
+                "",
+                xy=(x1, y1),
+                xytext=(x0, y0),
+                arrowprops={"arrowstyle": "->", "lw": 3.2, "color": color, "shrinkA": 22, "shrinkB": 22},
+            )
+        for (a, b), lbl in zip(zip(path, path[1:]), labels):
+            x0, y0 = idx[a]
+            x1, y1 = idx[b]
+            mx, my = (x0 + x1) / 2, (y0 + y1) / 2
+            if x0 == x1:  # vertical move -> label to the side
+                ax.text(
+                    mx + 0.30,
+                    my,
+                    lbl,
+                    color=color,
+                    fontsize=9.5,
+                    ha="left",
+                    va="center",
+                    weight="bold",
+                    bbox={"boxstyle": "round,pad=0.2", "facecolor": "white", "edgecolor": "none", "alpha": 0.85},
+                )
+            else:  # horizontal move -> label above
+                ax.text(
+                    mx,
+                    my - 0.32,
+                    lbl,
+                    color=color,
+                    fontsize=9.5,
+                    ha="center",
+                    va="bottom",
+                    weight="bold",
+                    bbox={"boxstyle": "round,pad=0.2", "facecolor": "white", "edgecolor": "none", "alpha": 0.85},
+                )
+
+    # Red: tune MNS first, then TP -> walks into a coordinate-wise local optimum.
+    draw_path([(1, 8), (1, 16), (4, 16)], "#C0392B", ["tune MNS", "tune TP"])
+    # Green: tune TP first, then MNS -> reaches the measured global best.
+    draw_path([(1, 8), (2, 8), (2, 32)], "#2E7D32", ["tune TP", "tune MNS"])
+
+    # start / trap / best markers (explained by the legend below the grid)
+    sx, sy = idx[(1, 8)]
+    ax.scatter([sx], [sy], marker="o", s=210, facecolors="none", edgecolors="black", linewidths=2.2, zorder=5)
+    trap = (4, 16)
+    tx, ty = idx[trap]
+    ax.add_patch(plt.Rectangle((tx - 0.5, ty - 0.5), 1, 1, fill=False, edgecolor="#C0392B", linewidth=3.4))
+    best = (2, 32)
+    bx, by = idx[best]
+    ax.add_patch(plt.Rectangle((bx - 0.5, by - 0.5), 1, 1, fill=False, edgecolor="#2E7D32", linewidth=3.4))
+
+    legend_elements = [
+        Line2D([0], [0], marker="o", color="w", markerfacecolor="none", markeredgecolor="black", markeredgewidth=2, markersize=10, label="start  TP1,MNS8 = 2.10"),
+        Patch(facecolor="none", edgecolor="#C0392B", linewidth=2.4, label="OAT trap  TP4,MNS16 = 2.44  (no improving single-knob move)"),
+        Patch(facecolor="none", edgecolor="#2E7D32", linewidth=2.4, label="global best  TP2,MNS32 = 3.28"),
+    ]
+    ax.legend(handles=legend_elements, loc="upper center", bbox_to_anchor=(0.5, -0.11), ncol=1, frameon=False, fontsize=9.5)
+    cbar = fig.colorbar(im, ax=ax, fraction=0.038, pad=0.035)
+    cbar.set_label("req/s/GPU")
+
+    # ---- right panel: why the red path stops (trap neighbourhood zoom) ----
+    ax2 = axes[1]
+    ax2.set_xlim(-0.55, 3.55)
+    ax2.set_ylim(-0.65, 3.7)
+    ax2.set_aspect("equal")
+    ax2.set_axis_off()
+    ax2.set_title("Why the red path stops here", fontsize=12, loc="left")
+
+    # 3x3 neighbourhood of the trap (TP4,MNS16). Rows match left panel:
+    # top=TP2, middle=TP4(trap), bottom=TP8(not measured). Cols: MNS 8/16/32.
+    # (col, row) with row 0 at bottom.
+    zoom_cells = {
+        (1, 2): ("2.275", "TP2", "dead"),   # up neighbour
+        (0, 1): ("1.28", "TP4", "dead"),    # left neighbour
+        (1, 1): ("2.44", "TP4", "trap"),    # the trap
+        (2, 1): ("2.44", "TP4", "flat"),    # right neighbour (flat, not strictly improving)
+        (1, 0): ("—", "TP8", "oob"),        # down neighbour (not measured)
+    }
+    for (col, row), (val, tplabel, kind) in zoom_cells.items():
+        x, y = col, row
+        if kind == "trap":
+            fc, ec, lw = "#FDECEA", "#C0392B", 3.2
+        elif kind == "oob":
+            fc, ec, lw = "#F2F2F2", "#CCCCCC", 1.0
+        else:
+            fc, ec, lw = "#FDF2F2", "#E6B0AA", 1.4
+        ax2.add_patch(plt.Rectangle((x, y), 1, 1, facecolor=fc, edgecolor=ec, linewidth=lw))
+        if kind == "oob":
+            ax2.text(x + 0.5, y + 0.5, "no data", ha="center", va="center", fontsize=9, color="#999", style="italic")
+        else:
+            val_color = "#C0392B" if kind in ("dead", "flat") else "#222"
+            ax2.text(x + 0.5, y + 0.62, val, ha="center", va="center", fontsize=13, weight="bold", color=val_color)
+            sublabel = tplabel
+            if kind == "trap":
+                sublabel = f"{tplabel} · trap"
+            elif kind == "flat":
+                sublabel = f"{tplabel} · flat"
+            ax2.text(x + 0.5, y + 0.28, sublabel, ha="center", va="center", fontsize=8.5, color="#666")
+        if kind in ("dead", "flat"):
+            ax2.text(x + 0.87, y + 0.87, "✗", ha="center", va="center", color="#C0392B", fontsize=18, weight="bold")
+
+    # axis labels for the zoom
+    ax2.text(-0.18, 2.5, "TP=2", ha="right", va="center", fontsize=9, color="#666")
+    ax2.text(-0.18, 1.5, "TP=4", ha="right", va="center", fontsize=9, color="#666")
+    ax2.text(-0.18, 0.5, "TP=8", ha="right", va="center", fontsize=9, color="#999")
+    ax2.text(0.5, -0.18, "MNS=8", ha="center", va="top", fontsize=9, color="#666")
+    ax2.text(1.5, -0.18, "MNS=16", ha="center", va="top", fontsize=9, color="#666")
+    ax2.text(2.5, -0.18, "MNS=32", ha="center", va="top", fontsize=9, color="#666")
+
+    ax2.text(
+        1.5,
+        3.45,
+        "Every measured single-knob move from the trap\nis worse or flat → coordinate ascent is stuck",
+        ha="center",
+        va="center",
+        fontsize=10.5,
+        color="#222",
+        bbox={"boxstyle": "round,pad=0.35", "facecolor": "white", "edgecolor": "#C0392B"},
+    )
+
+    fig.suptitle("One-knob-at-a-time tuning gets trapped: 25.6% throughput gap between two tuning orders", fontsize=14, y=1.02)
+    fig.tight_layout()
+    save(fig, "knob-oat-counterexample-c1-qwen30b")
+    plt.close(fig)
+
+
+def plot_c1_interaction_residual() -> None:
+    # If TP and MNS were independent additive effects, this residual matrix would be near zero.
+    mns = [8, 16, 32, 64]
+    tp = [1, 2, 4]
+    values = np.array(
+        [
+            [2.1000, 2.3500, 2.2833, 2.2833],
+            [2.2750, 2.2750, 3.2833, 3.2583],
+            [1.2833, 2.4417, 2.4417, 2.4417],
+        ]
+    )
+    residual = values - values.mean(axis=1, keepdims=True) - values.mean(axis=0, keepdims=True) + values.mean()
+    fig, ax = plt.subplots(figsize=(7.2, 4.8))
+    limit = float(np.abs(residual).max())
+    im = ax.imshow(residual, cmap="RdBu", aspect="auto", vmin=-limit, vmax=limit)
+    ax.set_xticks(range(len(mns)), labels=mns)
+    ax.set_yticks(range(len(tp)), labels=[f"TP={x}" for x in tp])
+    ax.set_xlabel("max-num-seqs (MNS)")
+    ax.set_ylabel("tensor-parallel-size")
+    ax.set_title("C1 non-additive interaction residual")
+    for i in range(residual.shape[0]):
+        for j in range(residual.shape[1]):
+            ax.text(j, i, f"{residual[i, j]:+.2f}", ha="center", va="center", fontsize=10)
+    cbar = fig.colorbar(im, ax=ax, fraction=0.046, pad=0.04)
+    cbar.set_label("req/s/GPU residual")
+    fig.suptitle("Independent-knob additive model leaves large structured residuals", fontsize=13, y=1.02)
+    fig.tight_layout()
+    save(fig, "knob-interaction-residual-c1-qwen30b")
+    plt.close(fig)
+
+
+def plot_c3_lines() -> None:
+    # Qwen235B decode C3, topology x MNS x MBT screen.
+    # Source run:
+    # interaction-qwen235b-decode-c3-topo-mns-mbt-fixed-dash1-d8899c5-20260703T022514Z
+    data = {
+        ("TP4/DP2/EP8", 64, 256): 0.05354166666666667,
+        ("TP4/DP2/EP8", 64, 384): 0.05354166666666667,
+        ("TP4/DP2/EP8", 128, 256): 0.058958333333333335,
+        ("TP4/DP2/EP8", 128, 384): 0.058958333333333335,
+        ("TP2/DP4/EP8", 64, 256): 0.058958333333333335,
+        ("TP2/DP4/EP8", 64, 384): 0.05354166666666667,
+        ("TP2/DP4/EP8", 128, 256): 0.058958333333333335,
+        ("TP2/DP4/EP8", 128, 384): 0.058958333333333335,
+    }
+    mbt = [256, 384]
+    topologies = ["TP4/DP2/EP8", "TP2/DP4/EP8"]
+
+    fig, axes = plt.subplots(1, 2, figsize=(11.5, 4.6), sharey=True)
+    for ax, topo in zip(axes, topologies):
+        for mns, color in [(64, "#4E79A7"), (128, "#F28E2B")]:
+            vals = [data[(topo, mns, b)] for b in mbt]
+            ax.plot(mbt, vals, marker="o", linewidth=2.6, color=color, label=f"MNS={mns}")
+            for x, y in zip(mbt, vals):
+                ax.text(x, y + 0.0007, f"{y:.4f}", ha="center", fontsize=9)
+        ax.set_title(topo)
+        ax.set_xlabel("max-num-batched-tokens (MBT)")
+        ax.set_xticks(mbt)
+        ax.grid(True, axis="y", alpha=0.28)
+        ax.set_ylim(0.050, 0.062)
+    axes[0].set_ylabel("req/s/GPU")
+    axes[1].legend(frameon=False, loc="lower right")
+    fig.suptitle("C3: MBT effect depends on topology and MNS", fontsize=14, y=1.02)
+    fig.tight_layout()
+    save(fig, "knob-conditional-c3-qwen235b-decode-lines")
+    plt.close(fig)
+
+
+def plot_delta_summary() -> None:
+    c1_base = {
+        "TP=1": (2.2833 - 2.1000) / 2.1000 * 100.0,
+        "TP=2": (3.2833 - 2.2750) / 2.2750 * 100.0,
+        "TP=4": (2.4417 - 1.2833) / 1.2833 * 100.0,
+    }
+    c3_mbt = {
+        "TP4/DP2\nMNS=64": 0.0,
+        "TP4/DP2\nMNS=128": 0.0,
+        "TP2/DP4\nMNS=64": (0.05354166666666667 - 0.058958333333333335)
+        / 0.058958333333333335
+        * 100.0,
+        "TP2/DP4\nMNS=128": 0.0,
+    }
+    c3_mns = {
+        "TP4/DP2\nMBT=256": (0.058958333333333335 - 0.05354166666666667)
+        / 0.05354166666666667
+        * 100.0,
+        "TP4/DP2\nMBT=384": (0.058958333333333335 - 0.05354166666666667)
+        / 0.05354166666666667
+        * 100.0,
+        "TP2/DP4\nMBT=256": 0.0,
+        "TP2/DP4\nMBT=384": (0.058958333333333335 - 0.05354166666666667)
+        / 0.05354166666666667
+        * 100.0,
+    }
+
+    panels = [
+        ("C1: MNS 8->32\nunder different TP", c1_base, "#59A14F"),
+        ("C3: MBT 256->384\nunder different context", c3_mbt, "#E15759"),
+        ("C3: MNS 64->128\nunder different context", c3_mns, "#4E79A7"),
+    ]
+    fig, axes = plt.subplots(1, 3, figsize=(16, 5.2))
+    for ax, (title, vals, color) in zip(axes, panels):
+        labels = list(vals.keys())
+        y = np.arange(len(labels))
+        x = list(vals.values())
+        colors = [color if v >= 0 else "#B07AA1" for v in x]
+        ax.barh(y, x, color=colors)
+        ax.axvline(0, color="black", linewidth=0.8)
+        lo = min(x)
+        hi = max(x)
+        pad = max(2.0, (hi - lo) * 0.12)
+        ax.set_xlim(lo - pad, hi + pad)
+        ax.set_yticks(y, labels=labels, fontsize=8)
+        ax.invert_yaxis()
+        ax.set_xlabel("relative change in req/s/GPU (%)")
+        ax.set_title(title)
+        ax.grid(True, axis="x", alpha=0.25)
+        for yi, xi in zip(y, x):
+            ha = "left" if xi >= 0 else "right"
+            offset = 0.7 if xi >= 0 else -0.7
+            ax.text(xi + offset, yi, f"{xi:+.1f}%", va="center", ha=ha, fontsize=8)
+    fig.suptitle("The same knob intervention has context-dependent effect size", fontsize=14, y=1.02)
+    fig.tight_layout()
+    save(fig, "knob-conditional-delta-summary")
+    plt.close(fig)
+
+
+def main() -> None:
+    plot_c1_oat_counterexample()
+    plot_c1_interaction_residual()
+    plot_c1_surface()
+    plot_c3_lines()
+    plot_delta_summary()
+
+
+if __name__ == "__main__":
+    main()

src/aituner/cli.py

@@ -1,17 +1,24 @@
 from __future__ import annotations
 
 import argparse
+import hashlib
 import json
 import sys
 from dataclasses import replace
 from pathlib import Path
+from typing import Any
 
 from .compare import run_compare
+from .config_signature import (
+    materialized_effective_config_signature,
+    tested_config_signature_index,
+)
 from .harness import (
     build_harness_context,
     build_harness_guided_proposal,
     build_harness_stop_proposal,
 )
+from .interaction_matrix import build_interaction_screening_matrix
 from .job import append_job, build_trial_job
 from .lca import (
     build_study_workload_profile,
@@ -19,11 +26,21 @@ from .lca import (
     resolve_length_mode,
     similarity_report,
 )
-from .llm import build_prompt, call_llm_for_proposal, load_capability_profile, parse_proposal_text
+from .llm import (
+    build_initial_config_review_prompt,
+    build_prompt,
+    call_llm_for_initial_config_review,
+    call_llm_for_proposal,
+    load_capability_profile,
+    parse_initial_config_review_text,
+    parse_proposal_text,
+)
 from .spec import (
+    ConfigPatch,
     Proposal,
     SpecError,
     StudySpec,
+    StudyState,
     load_structured_file,
     load_study_spec,
     to_jsonable,
@@ -37,6 +54,435 @@ def _is_empty_config_patch(proposal: Proposal) -> bool:
     return not proposal.config_patch.env_patch and not proposal.config_patch.flag_patch
 
 
+def _reject_repeated_effective_config(
+    *,
+    study: StudySpec,
+    state: StudyState,
+    proposal: Proposal,
+    proposal_name: str,
+) -> None:
+    if proposal.should_stop:
+        return
+    tested = tested_config_signature_index(study, state)
+    signature = materialized_effective_config_signature(
+        study=study,
+        state=state,
+        proposal=proposal,
+    )
+    matching_trials = tested.get(signature)
+    if not matching_trials:
+        return
+    raise SpecError(
+        f"Proposal {proposal_name} repeats an already tested effective full config "
+        "after materialization. "
+        f"matching_trial_ids={matching_trials}. "
+        "Choose a different eligible candidate or return should_stop=true."
+    )
+
+
+def _effective_config_fingerprint(signature: str) -> str:
+    return hashlib.sha256(signature.encode("utf-8")).hexdigest()
+
+
+def _proposal_effective_config_fingerprint(
+    *,
+    study: StudySpec,
+    state: StudyState,
+    proposal: Proposal,
+) -> str | None:
+    if proposal.should_stop:
+        return None
+    signature = materialized_effective_config_signature(
+        study=study,
+        state=state,
+        proposal=proposal,
+    )
+    return _effective_config_fingerprint(signature)
+
+
+def _visible_harness_candidates(
+    harness_context: dict[str, object] | None,
+) -> list[dict[str, Any]]:
+    if not isinstance(harness_context, dict):
+        return []
+    experiment_plan = harness_context.get("experiment_plan")
+    if not isinstance(experiment_plan, dict):
+        return []
+    candidate_set = experiment_plan.get("candidate_set")
+    if not isinstance(candidate_set, dict):
+        return []
+    candidates = candidate_set.get("eligible_candidates")
+    if not isinstance(candidates, list):
+        return []
+    return [item for item in candidates if isinstance(item, dict)]
+
+
+def _candidate_set_hash(harness_context: dict[str, object] | None) -> object:
+    if not isinstance(harness_context, dict):
+        return None
+    experiment_plan = harness_context.get("experiment_plan")
+    if not isinstance(experiment_plan, dict):
+        return None
+    candidate_set = experiment_plan.get("candidate_set")
+    if not isinstance(candidate_set, dict):
+        return None
+    return candidate_set.get("candidate_set_hash")
+
+
+def _match_visible_harness_candidate(
+    *,
+    proposal_fingerprint: str | None,
+    harness_context: dict[str, object] | None,
+) -> dict[str, Any] | None:
+    if proposal_fingerprint is None:
+        return None
+    for candidate in _visible_harness_candidates(harness_context):
+        if candidate.get("effective_config_fingerprint") == proposal_fingerprint:
+            return candidate
+    return None
+
+
+def _proposal_source_label(
+    *,
+    proposal_name: str,
+    proposal_source: Path | None,
+) -> str:
+    if proposal_name.startswith("baseline-"):
+        return "baseline"
+    if proposal_name.startswith("harness-stop-") or proposal_name.startswith("harness-proposal-"):
+        return "harness"
+    return str(proposal_source) if proposal_source else "llm"
+
+
+def _classify_proposal_attribution(
+    *,
+    study: StudySpec,
+    state: StudyState,
+    proposal: Proposal,
+    proposal_name: str,
+    proposal_source: Path | None,
+    harness_context: dict[str, object] | None,
+) -> dict[str, Any]:
+    source_label = _proposal_source_label(
+        proposal_name=proposal_name,
+        proposal_source=proposal_source,
+    )
+    fingerprint = _proposal_effective_config_fingerprint(
+        study=study,
+        state=state,
+        proposal=proposal,
+    )
+    matched = _match_visible_harness_candidate(
+        proposal_fingerprint=fingerprint,
+        harness_context=harness_context,
+    )
+    matched_candidate_id = matched.get("candidate_id") if matched else None
+    if proposal.should_stop:
+        origin = (
+            "harness_stop"
+            if proposal_name.startswith("harness-stop-")
+            else "proposal_file_stop"
+            if proposal_source is not None
+            else "llm_stop"
+        )
+    elif proposal_name.startswith("baseline-"):
+        origin = "baseline"
+    elif proposal_name.startswith("harness-proposal-"):
+        origin = "harness_top1"
+    elif proposal_source is not None:
+        origin = "proposal_file"
+    elif study.llm.use_harness and harness_context is not None:
+        origin = "llm_selected_harness_candidate" if matched else "llm_out_of_set"
+    else:
+        origin = "llm_no_harness"
+    return {
+        "schema_version": 1,
+        "proposal_name": proposal_name,
+        "proposal_source": source_label,
+        "proposal_origin": origin,
+        "harness_candidate_policy": study.llm.harness_candidate_policy,
+        "candidate_set_hash": _candidate_set_hash(harness_context),
+        "proposal_effective_config_fingerprint": fingerprint,
+        "matched_effective_config_signature": matched is not None,
+        "matched_candidate_id": matched_candidate_id,
+        "matched_candidate": _compact_candidate_for_attribution(matched),
+    }
+
+
+def _compact_candidate_for_attribution(candidate: dict[str, Any] | None) -> dict[str, Any] | None:
+    if not candidate:
+        return None
+    return {
+        "candidate_id": candidate.get("candidate_id"),
+        "action_id": candidate.get("action_id"),
+        "knob_family": candidate.get("knob_family"),
+        "score": candidate.get("score"),
+        "config_patch": candidate.get("config_patch"),
+    }
+
+
+def _validate_harness_candidate_policy(attribution: dict[str, Any]) -> None:
+    if (
+        attribution.get("harness_candidate_policy") == "strict"
+        and attribution.get("proposal_origin") == "llm_out_of_set"
+    ):
+        raise SpecError(
+            "LLM proposal is outside the visible harness eligible candidate set while "
+            "llm.harness_candidate_policy=strict. Use an eligible harness candidate, "
+            "switch to advisory mode, or disable harness context."
+        )
+
+
+def _config_patch_knob_keys(config_patch: object) -> set[str]:
+    if not isinstance(config_patch, dict):
+        return set()
+    keys: set[str] = set()
+    env_patch = config_patch.get("env_patch")
+    if isinstance(env_patch, dict):
+        keys.update(f"env:{key}" for key in env_patch)
+    flag_patch = config_patch.get("flag_patch")
+    if isinstance(flag_patch, dict):
+        keys.update(f"flag:{key}" for key in flag_patch)
+    return keys
+
+
+def _proposal_knob_keys(proposal: Proposal) -> set[str]:
+    return {
+        *{f"env:{key}" for key in proposal.config_patch.env_patch},
+        *{f"flag:{key}" for key in proposal.config_patch.flag_patch},
+    }
+
+
+def _nearest_visible_harness_candidates(
+    *,
+    proposal: Proposal,
+    harness_context: dict[str, object] | None,
+    limit: int = 3,
+) -> list[dict[str, Any]]:
+    proposal_keys = _proposal_knob_keys(proposal)
+    scored: list[tuple[int, float, dict[str, Any]]] = []
+    for candidate in _visible_harness_candidates(harness_context):
+        candidate_keys = _config_patch_knob_keys(candidate.get("config_patch"))
+        overlap = len(proposal_keys & candidate_keys)
+        score = candidate.get("score")
+        candidate_score = float(score) if isinstance(score, (int, float)) else 0.0
+        scored.append((overlap, candidate_score, candidate))
+    scored.sort(key=lambda item: (item[0], item[1]), reverse=True)
+    return [_compact_candidate_for_attribution(item[2]) or {} for item in scored[:limit]]
+
+
+def _candidate_family_gap_payload(
+    *,
+    study: StudySpec,
+    trial_id: str,
+    proposal_name: str,
+    proposal: Proposal,
+    attribution: dict[str, Any],
+    harness_context: dict[str, object] | None,
+    incumbent_rate_per_gpu: float,
+    result_rate_per_gpu: float,
+) -> dict[str, Any]:
+    nearest = _nearest_visible_harness_candidates(
+        proposal=proposal,
+        harness_context=harness_context,
+    )
+    proposal_keys = _proposal_knob_keys(proposal)
+    has_same_knob_candidate = any(
+        proposal_keys
+        & _config_patch_knob_keys(candidate.get("config_patch") if isinstance(candidate, dict) else None)
+        for candidate in nearest
+    )
+    return {
+        "schema_version": 1,
+        "study_id": study.study_id,
+        "trial_id": trial_id,
+        "proposal_name": proposal_name,
+        "proposal_origin": attribution.get("proposal_origin"),
+        "gap_type": "same_operator_new_step" if has_same_knob_candidate else "missing_operator",
+        "review_status": "pending",
+        "incumbent_request_rate_per_gpu": incumbent_rate_per_gpu,
+        "result_request_rate_per_gpu": result_rate_per_gpu,
+        "absolute_gain": result_rate_per_gpu - incumbent_rate_per_gpu,
+        "relative_gain": (
+            (result_rate_per_gpu - incumbent_rate_per_gpu) / incumbent_rate_per_gpu
+            if incumbent_rate_per_gpu > 0
+            else None
+        ),
+        "proposal_patch": {
+            "env_patch": dict(proposal.config_patch.env_patch),
+            "flag_patch": dict(proposal.config_patch.flag_patch),
+        },
+        "changed_knobs": sorted(proposal_keys),
+        "candidate_set_hash": attribution.get("candidate_set_hash"),
+        "nearest_harness_candidates": nearest,
+        "interpretation": (
+            "LLM changed a knob already present in the visible harness candidate set; "
+            "treat this as a step-size/acquisition gap until offline review accepts "
+            "a descriptor or operator change."
+            if has_same_knob_candidate
+            else "LLM changed knobs not represented by the visible harness candidates; "
+            "offline review should decide whether this is a missing operator, "
+            "descriptor, or mechanism."
+        ),
+    }
+
+
+def _result_request_rate_per_gpu(result: dict[str, object], parallel_size: int | None) -> float | None:
+    best_request_rate = result.get("best_request_rate")
+    if not isinstance(best_request_rate, (int, float)):
+        return None
+    if not isinstance(parallel_size, int) or parallel_size <= 0:
+        return None
+    return float(best_request_rate) / float(parallel_size)
+
+
+def _parse_parallel_int(value: object, *, default: int = 1) -> int:
+    if value is None:
+        return default
+    if isinstance(value, bool):
+        raise SpecError("Boolean values are not valid topology settings.")
+    if isinstance(value, int):
+        return value
+    if isinstance(value, float) and value.is_integer():
+        return int(value)
+    if isinstance(value, str) and value.strip():
+        return int(value.strip())
+    raise SpecError(f"Unable to parse topology setting from {value!r}.")
+
+
+def _parallel_size_for_config_patch(study: StudySpec, config_patch: object) -> int | None:
+    if not isinstance(config_patch, ConfigPatch):
+        return None
+    flags: dict[str, object] = dict(study.engine.base_flags)
+    flags.update(config_patch.flag_patch)
+    tp = _parse_parallel_int(flags.get("tensor-parallel-size"), default=1)
+    dp = _parse_parallel_int(flags.get("data-parallel-size"), default=1)
+    return tp * dp
+
+
+def _is_candidate_family_gap(
+    *,
+    attribution: dict[str, Any],
+    incumbent_rate_per_gpu: float | None,
+    result_rate_per_gpu: float | None,
+) -> bool:
+    if attribution.get("proposal_origin") != "llm_out_of_set":
+        return False
+    if not isinstance(incumbent_rate_per_gpu, (int, float)):
+        return False
+    if not isinstance(result_rate_per_gpu, (int, float)):
+        return False
+    min_gain = max(1e-6, float(incumbent_rate_per_gpu) * 0.01)
+    return float(result_rate_per_gpu) > float(incumbent_rate_per_gpu) + min_gain
+
+
+def _harness_snapshot_payload(
+    *,
+    study: StudySpec,
+    state: StudyState,
+    harness_context: dict[str, object],
+) -> dict[str, object]:
+    experiment_plan = harness_context.get("experiment_plan")
+    if not isinstance(experiment_plan, dict):
+        experiment_plan = {}
+    candidate_set = experiment_plan.get("candidate_set")
+    if not isinstance(candidate_set, dict):
+        candidate_set = {}
+    return {
+        "schema_version": 1,
+        "study_id": study.study_id,
+        "iteration": state.next_trial_index,
+        "planner_version": experiment_plan.get("planner_version"),
+        "candidate_set_hash": candidate_set.get("candidate_set_hash"),
+        "state_ref": {
+            "best_trial_id": state.best_trial_id,
+            "best_parallel_size": state.best_parallel_size,
+            "best_request_rate": state.best_request_rate,
+            "best_request_rate_per_gpu": state.best_request_rate_per_gpu,
+            "next_trial_index": state.next_trial_index,
+            "trial_count": len(state.trials),
+        },
+        "candidate_set": candidate_set,
+        "decisions": {
+            "next_action": experiment_plan.get("next_action"),
+            "harness_proposal": harness_context.get("harness_proposal"),
+            "harness_stop": harness_context.get("harness_stop"),
+            "stop_authority": harness_context.get("stop_authority"),
+        },
+    }
+
+
+def _maybe_run_initial_config_review(
+    *,
+    study: StudySpec,
+    spec_path: Path,
+    store: StudyStore,
+    capability_profile: dict[str, Any] | None,
+) -> dict[str, Any] | None:
+    mode = study.llm.initial_config_review.mode
+    if mode == "off":
+        return None
+    state = store.load_state(study.study_id)
+    if state.trials or state.next_trial_index != 1:
+        return None
+
+    audit_name = "initial-config-0001"
+    audit_root = store.study_root(study.study_id) / "preflight_audits"
+    audit_path = audit_root / f"{audit_name}.json"
+    if audit_path.exists():
+        return json.loads(audit_path.read_text(encoding="utf-8"))
+
+    base_payload: dict[str, Any] = {
+        "schema_version": 1,
+        "audit_type": "initial_config_review",
+        "study_id": study.study_id,
+        "mode": mode,
+        "repair_applied": False,
+    }
+    if study.llm.endpoint is None:
+        payload = {
+            **base_payload,
+            "status": "skipped",
+            "reason": "llm.endpoint_not_configured",
+        }
+        store.write_preflight_audit(study.study_id, audit_name, payload)
+        return payload
+
+    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)
+    prompt = build_initial_config_review_prompt(
+        study=study,
+        window_summary=window_summary,
+        capability_profile=capability_profile,
+        workload_profile=workload_profile,
+    )
+    prompt_path = audit_root / f"{audit_name}.prompt.txt"
+    raw_path = audit_root / f"{audit_name}.raw.txt"
+    prompt_path.write_text(prompt, encoding="utf-8")
+    try:
+        raw_text = call_llm_for_initial_config_review(policy=study.llm, prompt=prompt)
+        raw_path.write_text(raw_text, encoding="utf-8")
+        review = parse_initial_config_review_text(raw_text, study)
+        payload = {
+            **base_payload,
+            "status": "completed",
+            "prompt_path": str(prompt_path),
+            "raw_response_path": str(raw_path),
+            "review": review,
+        }
+    except Exception as exc:  # noqa: BLE001
+        payload = {
+            **base_payload,
+            "status": "failed",
+            "prompt_path": str(prompt_path),
+            "raw_response_path": str(raw_path) if raw_path.exists() else None,
+            "error": str(exc),
+        }
+    store.write_preflight_audit(study.study_id, audit_name, payload)
+    return payload
+
+
 def _latency_percentiles(summary: object, metric: str) -> dict[str, float]:
     if not isinstance(summary, dict):
         return {}
@@ -219,8 +665,15 @@ def cmd_study_tune(args: argparse.Namespace) -> int:
     store = StudyStore(Path(args.store_root) if args.store_root else None)
     study_root = store.init_study(spec_path=spec_path, study=study)
     capability_profile = load_capability_profile(study, study_spec_path=spec_path)
+    preflight_audit = _maybe_run_initial_config_review(
+        study=study,
+        spec_path=spec_path,
+        store=store,
+        capability_profile=capability_profile,
+    )
     proposal_files = [Path(item).resolve() for item in (args.proposal_file or [])]
     max_trials = args.max_trials or (len(proposal_files) if proposal_files else 2)
+    proposal_policy = args.proposal_policy
     if max_trials <= 0:
         raise SpecError("max_trials must be positive")
     if proposal_files and max_trials > len(proposal_files):
@@ -258,6 +711,16 @@ def cmd_study_tune(args: argparse.Namespace) -> int:
             if study.llm.use_harness
             else None
         )
+        if harness_context is not None:
+            store.write_harness_snapshot(
+                study.study_id,
+                f"candidate-set-{state.next_trial_index:04d}",
+                _harness_snapshot_payload(
+                    study=study,
+                    state=state,
+                    harness_context=harness_context,
+                ),
+            )
         prompt = build_prompt(
             study=study,
             window_summary=window_summary,
@@ -310,7 +773,7 @@ def cmd_study_tune(args: argparse.Namespace) -> int:
             else:
                 guided_proposal = (
                     build_harness_guided_proposal(harness_context)
-                    if harness_context is not None
+                    if harness_context is not None and proposal_policy == "harness-first"
                     else None
                 )
                 if guided_proposal is not None:
@@ -334,8 +797,29 @@ def cmd_study_tune(args: argparse.Namespace) -> int:
         raw_proposal_path = store.study_root(study.study_id) / "proposals" / f"{proposal_name}.raw.txt"
         raw_proposal_path.write_text(proposal_text, encoding="utf-8")
         proposal = parse_proposal_text(proposal_text, study)
+        _reject_repeated_effective_config(
+            study=study,
+            state=state,
+            proposal=proposal,
+            proposal_name=proposal_name,
+        )
+        proposal_attribution = _classify_proposal_attribution(
+            study=study,
+            state=state,
+            proposal=proposal,
+            proposal_name=proposal_name,
+            proposal_source=proposal_source,
+            harness_context=harness_context,
+        )
+        _validate_harness_candidate_policy(proposal_attribution)
         store.write_proposal(study.study_id, proposal_name, proposal)
         if proposal.should_stop:
+            proposal_attribution["stopped"] = True
+            store.write_proposal_attribution(
+                study.study_id,
+                proposal_name,
+                proposal_attribution,
+            )
             is_harness_stop = proposal_name.startswith("harness-stop-")
             is_llm_stop = not is_harness_stop and proposal_source is None
             stop_authority = (
@@ -414,21 +898,55 @@ def cmd_study_tune(args: argparse.Namespace) -> int:
             and not state.trials
             and _is_empty_config_patch(proposal)
         )
+        pre_trial_best_rate_per_gpu = state.best_request_rate_per_gpu
         trial, _ = store.materialize_trial(study=study, state=state, proposal=proposal)
+        proposal_attribution["trial_id"] = trial.trial_id
+        store.write_proposal_attribution(
+            study.study_id,
+            proposal_name,
+            proposal_attribution,
+        )
         trial_spec_path = Path(trial.artifact_dir) / "trial_spec.json"
         result = run_trial(trial_spec_path)
         state = store.ingest_trial_results(study.study_id)
+        trial_parallel_size = _parallel_size_for_config_patch(study, trial.config_patch)
+        result_rate_per_gpu = _result_request_rate_per_gpu(result, trial_parallel_size)
+        gap_path: Path | None = None
+        if _is_candidate_family_gap(
+            attribution=proposal_attribution,
+            incumbent_rate_per_gpu=pre_trial_best_rate_per_gpu,
+            result_rate_per_gpu=result_rate_per_gpu,
+        ):
+            gap_payload = _candidate_family_gap_payload(
+                study=study,
+                trial_id=trial.trial_id,
+                proposal_name=proposal_name,
+                proposal=proposal,
+                attribution=proposal_attribution,
+                harness_context=harness_context,
+                incumbent_rate_per_gpu=float(pre_trial_best_rate_per_gpu),
+                result_rate_per_gpu=float(result_rate_per_gpu),
+            )
+            gap_path = store.write_candidate_family_gap(
+                study.study_id,
+                trial.trial_id,
+                gap_payload,
+            )
         executed.append(
-                {
+            {
-                    "trial_id": trial.trial_id,
+                "trial_id": trial.trial_id,
-                    "proposal_name": proposal_name,
+                "proposal_name": proposal_name,
-                    "proposal_source": (
+                "proposal_source": (
-                        "harness"
+                    "harness"
-                        if proposal_name.startswith("harness-proposal-")
+                    if proposal_name.startswith("harness-proposal-")
-                        else str(proposal_source) if proposal_source else "llm"
+                    else str(proposal_source) if proposal_source else "llm"
-                    ),
+                ),
+                "proposal_origin": proposal_attribution.get("proposal_origin"),
+                "matched_candidate_id": proposal_attribution.get("matched_candidate_id"),
+                "candidate_family_gap_path": str(gap_path) if gap_path else None,
                 "best_sampling_u": result.get("best_sampling_u"),
                 "best_request_rate": result.get("best_request_rate"),
+                "best_request_rate_per_gpu": result_rate_per_gpu,
                 "best_pass_rate": result.get("best_pass_rate"),
                 "state_best_trial_id": state.best_trial_id,
                 "state_best_request_rate": state.best_request_rate,
@@ -460,6 +978,7 @@ def cmd_study_tune(args: argparse.Namespace) -> int:
         json.dumps(
             {
                 "study_root": str(study_root),
+                "preflight_audit": preflight_audit,
                 "executed_trials": executed,
                 "best_trial_id": final_state.best_trial_id,
                 "best_request_rate": final_state.best_request_rate,
@@ -651,6 +1170,18 @@ def cmd_profile_similarity(args: argparse.Namespace) -> int:
     return 0
 
 
+def cmd_profile_interaction_matrix(args: argparse.Namespace) -> int:
+    spec_path = Path(args.spec).resolve()
+    study = _load_profile_study_spec(spec_path)
+    window, requests = load_trace_requests(study, study_spec_path=spec_path)
+    matrix = build_interaction_screening_matrix(
+        study=study,
+        window_summary=summarize_window(requests, window),
+    )
+    print(json.dumps(matrix, ensure_ascii=False, indent=2))
+    return 0
+
+
 def build_parser() -> argparse.ArgumentParser:
     parser = argparse.ArgumentParser(description="AITuner CLI")
     subparsers = parser.add_subparsers(dest="command", required=True)
@@ -699,6 +1230,15 @@ def build_parser() -> argparse.ArgumentParser:
     tune.add_argument("--store-root")
     tune.add_argument("--proposal-file", action="append")
     tune.add_argument("--max-trials", type=int)
+    tune.add_argument(
+        "--proposal-policy",
+        choices=("harness-first", "llm-first"),
+        default="harness-first",
+        help=(
+            "Choose whether deterministic harness proposals are tried before the LLM "
+            "or whether the LLM proposes directly from the harness prompt/context."
+        ),
+    )
     tune.add_argument(
         "--skip-baseline",
         action="store_true",
@@ -763,6 +1303,10 @@ def build_parser() -> argparse.ArgumentParser:
     )
     profile_similarity.set_defaults(func=cmd_profile_similarity)
 
+    profile_interaction = profile_sub.add_parser("interaction-matrix")
+    profile_interaction.add_argument("--spec", required=True)
+    profile_interaction.set_defaults(func=cmd_profile_interaction_matrix)
+
     return parser
 
 

src/aituner/config_signature.py

@@ -0,0 +1,148 @@
+from __future__ import annotations
+
+import json
+from dataclasses import replace
+from typing import Any
+
+from .spec import ConfigPatch, Proposal, StudySpec, StudyState
+
+
+TOPOLOGY_FLAG_KEYS = {
+    "tensor-parallel-size",
+    "data-parallel-size",
+    "expert-parallel-size",
+    "enable-expert-parallel",
+}
+
+
+def normalized_config_patch(config_patch: Any) -> dict[str, dict[str, Any]]:
+    if isinstance(config_patch, ConfigPatch):
+        env_patch: Any = config_patch.env_patch
+        flag_patch: Any = config_patch.flag_patch
+    elif isinstance(config_patch, dict):
+        env_patch = config_patch.get("env_patch")
+        flag_patch = config_patch.get("flag_patch")
+    else:
+        env_patch = {}
+        flag_patch = {}
+    return {
+        "env_patch": _canonical_env_map(env_patch if isinstance(env_patch, dict) else {}),
+        "flag_patch": _canonical_flag_map(flag_patch if isinstance(flag_patch, dict) else {}),
+    }
+
+
+def effective_config_signature(study: StudySpec, config_patch: Any) -> str:
+    patch = normalized_config_patch(config_patch)
+    payload = {
+        "env": _canonical_env_map({**study.engine.base_envs, **patch["env_patch"]}),
+        "flags": _canonical_flag_map({**study.engine.base_flags, **patch["flag_patch"]}),
+    }
+    return json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+
+
+def materialize_proposal_for_execution(
+    *,
+    study: StudySpec,
+    state: StudyState,
+    proposal: Proposal,
+) -> Proposal:
+    flag_patch = dict(proposal.config_patch.flag_patch)
+    env_patch = dict(proposal.config_patch.env_patch)
+    if not flag_patch and not env_patch:
+        return proposal
+    if TOPOLOGY_FLAG_KEYS.intersection(flag_patch):
+        return proposal
+    if not state.best_trial_id:
+        return proposal
+    incumbent = next(
+        (trial for trial in state.trials if trial.trial_id == state.best_trial_id),
+        None,
+    )
+    if incumbent is None or not isinstance(incumbent.config_patch, dict):
+        return proposal
+    incumbent_patch = incumbent.config_patch.get("flag_patch")
+    if not isinstance(incumbent_patch, dict):
+        return proposal
+    inherited_topology = {
+        key: value
+        for key, value in incumbent_patch.items()
+        if key in TOPOLOGY_FLAG_KEYS and study.engine.base_flags.get(key) != value
+    }
+    if not inherited_topology:
+        return proposal
+    merged_flag_patch = dict(inherited_topology)
+    merged_flag_patch.update(flag_patch)
+    return replace(
+        proposal,
+        config_patch=ConfigPatch(
+            env_patch=env_patch,
+            flag_patch=merged_flag_patch,
+        ),
+    )
+
+
+def materialized_effective_config_signature(
+    *,
+    study: StudySpec,
+    state: StudyState,
+    proposal: Proposal,
+) -> str:
+    materialized = materialize_proposal_for_execution(
+        study=study,
+        state=state,
+        proposal=proposal,
+    )
+    return effective_config_signature(study, materialized.config_patch)
+
+
+def tested_config_signature_index(study: StudySpec, state: StudyState) -> dict[str, list[str]]:
+    index: dict[str, list[str]] = {}
+    for trial in state.trials:
+        signature = effective_config_signature(study, trial.config_patch)
+        index.setdefault(signature, []).append(trial.trial_id)
+    return index
+
+
+def _canonical_env_map(payload: dict[str, Any]) -> dict[str, str]:
+    return {str(key): str(value) for key, value in payload.items()}
+
+
+def _canonical_flag_map(payload: dict[str, Any]) -> dict[str, Any]:
+    return {str(key): _canonical_flag_value(value) for key, value in payload.items()}
+
+
+def _canonical_flag_value(value: Any) -> Any:
+    if value is None or isinstance(value, bool):
+        return value
+    if isinstance(value, int):
+        return value
+    if isinstance(value, float):
+        return int(value) if value.is_integer() else value
+    if isinstance(value, str):
+        return _canonical_string_flag_value(value)
+    if isinstance(value, list):
+        return [_canonical_flag_value(item) for item in value]
+    if isinstance(value, tuple):
+        return [_canonical_flag_value(item) for item in value]
+    if isinstance(value, dict):
+        return {str(key): _canonical_flag_value(item) for key, item in value.items()}
+    return str(value)
+
+
+def _canonical_string_flag_value(value: str) -> Any:
+    stripped = value.strip()
+    if not stripped:
+        return stripped
+    try:
+        parsed_int = int(stripped, 10)
+    except ValueError:
+        pass
+    else:
+        return parsed_int
+    try:
+        parsed_float = float(stripped)
+    except ValueError:
+        return stripped
+    if parsed_float.is_integer():
+        return int(parsed_float)
+    return parsed_float

src/aituner/declarative_harness.py

@@ -0,0 +1,395 @@
+from __future__ import annotations
+
+"""Experimental declarative harness substrate.
+
+This module intentionally stays separate from the production harness while the
+coverage-relative design is being validated. It models a small, typed subset of
+the proposed intervention grammar: axes, generic operators, complete candidate
+sets, failure regions, and stop reports.
+"""
+
+import hashlib
+import json
+from dataclasses import dataclass
+from typing import Any, Literal, Mapping, Sequence
+
+
+AxisKind = Literal["ordered_lattice", "bounded_numeric"]
+OperatorKind = Literal["bracket", "step_up", "step_down", "jump_to_floor", "local_climb"]
+RegionRelation = Literal["eq", "ge", "le"]
+
+
+@dataclass(frozen=True)
+class AxisSpec:
+    name: str
+    kind: AxisKind
+    values: tuple[Any, ...] = ()
+    floor: float | None = None
+    ceiling: float | None = None
+    step: float | None = None
+
+    def validate(self) -> None:
+        if not self.name:
+            raise ValueError("axis name must be non-empty")
+        if self.kind == "ordered_lattice":
+            if not self.values:
+                raise ValueError(f"ordered lattice axis {self.name!r} needs values")
+            if len(set(_stable_token(value) for value in self.values)) != len(self.values):
+                raise ValueError(f"ordered lattice axis {self.name!r} has duplicate values")
+            return
+        if self.floor is None or self.ceiling is None:
+            raise ValueError(f"bounded numeric axis {self.name!r} needs floor and ceiling")
+        if self.floor > self.ceiling:
+            raise ValueError(f"bounded numeric axis {self.name!r} has floor above ceiling")
+        if self.step is None or self.step <= 0:
+            raise ValueError(f"bounded numeric axis {self.name!r} needs a positive step")
+
+
+@dataclass(frozen=True)
+class OperatorSpec:
+    name: str
+    axis: str
+    kind: OperatorKind
+    harness_priority: float = 0.0
+
+
+@dataclass(frozen=True)
+class CoverageUnit:
+    axis: str
+    operator: str
+    target: Any
+
+    @property
+    def unit_id(self) -> str:
+        return coverage_unit_id(self.axis, self.operator, self.target)
+
+
+@dataclass(frozen=True)
+class CandidateAction:
+    action_id: str
+    operator: str
+    axis: str
+    patch: Mapping[str, Any]
+    harness_priority: float
+    planner_score: float | None = None
+    backend_score: float | None = None
+    coverage_units: tuple[CoverageUnit, ...] = ()
+    source_value: Any = None
+    target_value: Any = None
+
+    @property
+    def signature(self) -> str:
+        return config_signature(self.patch)
+
+
+@dataclass(frozen=True)
+class BlockedCandidate:
+    candidate: CandidateAction
+    reason: str
+
+
+@dataclass(frozen=True)
+class FailureRegion:
+    axis: str
+    relation: RegionRelation
+    value: Any
+    reason: str = "prior_failure"
+
+    def contains(self, candidate: CandidateAction) -> bool:
+        if candidate.axis != self.axis:
+            return False
+        target = candidate.target_value
+        if self.relation == "eq":
+            return target == self.value
+        if self.relation == "ge":
+            return target >= self.value
+        if self.relation == "le":
+            return target <= self.value
+        raise ValueError(f"unknown region relation {self.relation!r}")
+
+
+@dataclass(frozen=True)
+class CoverageState:
+    tested_signatures: frozenset[str] = frozenset()
+    covered_unit_ids: frozenset[str] = frozenset()
+    failed_regions: tuple[FailureRegion, ...] = ()
+
+
+@dataclass(frozen=True)
+class HarnessPolicy:
+    operators: tuple[OperatorSpec, ...]
+    no_repeat: bool = True
+    required_coverage_unit_ids: frozenset[str] = frozenset()
+
+
+@dataclass(frozen=True)
+class CandidateSet:
+    eligible: tuple[CandidateAction, ...]
+    blocked: tuple[BlockedCandidate, ...]
+    candidate_set_hash: str
+
+
+@dataclass(frozen=True)
+class StopReport:
+    should_stop: bool
+    reason: str
+    candidate_set_hash: str
+    uncovered_unit_ids: tuple[str, ...] = ()
+    eligible_count: int = 0
+    blocked_count: int = 0
+
+
+def config_signature(patch: Mapping[str, Any]) -> str:
+    return json.dumps(dict(patch), sort_keys=True, separators=(",", ":"), ensure_ascii=False)
+
+
+def coverage_unit_id(axis: str, operator: str, target: Any) -> str:
+    target_text = json.dumps(target, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
+    return f"{axis}:{operator}:{target_text}"
+
+
+def ordered_lattice_failure_region(
+    axis: AxisSpec,
+    failed_value: Any,
+    *,
+    direction: Literal["up", "down", "exact"],
+    reason: str = "prior_failure",
+) -> FailureRegion:
+    axis.validate()
+    if axis.kind != "ordered_lattice":
+        raise ValueError("ordered_lattice_failure_region requires an ordered lattice axis")
+    if failed_value not in axis.values:
+        raise ValueError(f"{failed_value!r} is not in lattice axis {axis.name!r}")
+    if direction == "up":
+        return FailureRegion(axis=axis.name, relation="ge", value=failed_value, reason=reason)
+    if direction == "down":
+        return FailureRegion(axis=axis.name, relation="le", value=failed_value, reason=reason)
+    return FailureRegion(axis=axis.name, relation="eq", value=failed_value, reason=reason)
+
+
+def enumerate_candidate_set(
+    state: Mapping[str, Any],
+    axes: Sequence[AxisSpec],
+    policy: HarnessPolicy,
+    coverage_state: CoverageState | None = None,
+) -> CandidateSet:
+    coverage_state = coverage_state or CoverageState()
+    axis_by_name = {axis.name: axis for axis in axes}
+    for axis in axes:
+        axis.validate()
+
+    eligible: list[CandidateAction] = []
+    blocked: list[BlockedCandidate] = []
+    for operator in sorted(
+        policy.operators,
+        key=lambda item: (item.axis, item.name, item.kind),
+    ):
+        axis = axis_by_name.get(operator.axis)
+        if axis is None:
+            raise ValueError(f"operator {operator.name!r} references unknown axis {operator.axis!r}")
+        generated, generated_blocked = _generate_operator_actions(state, axis, operator)
+        blocked.extend(generated_blocked)
+        for candidate in generated:
+            reason = _blocking_reason(candidate, policy, coverage_state)
+            if reason is None:
+                eligible.append(candidate)
+            else:
+                blocked.append(BlockedCandidate(candidate=candidate, reason=reason))
+
+    eligible_tuple = tuple(sorted(eligible, key=_candidate_sort_key))
+    blocked_tuple = tuple(
+        sorted(blocked, key=lambda item: (_candidate_sort_key(item.candidate), item.reason))
+    )
+    return CandidateSet(
+        eligible=eligible_tuple,
+        blocked=blocked_tuple,
+        candidate_set_hash=_candidate_set_hash(eligible_tuple, blocked_tuple),
+    )
+
+
+def validate_coverage_stop(
+    candidate_set: CandidateSet,
+    policy: HarnessPolicy,
+    coverage_state: CoverageState,
+) -> StopReport:
+    uncovered = tuple(sorted(policy.required_coverage_unit_ids - coverage_state.covered_unit_ids))
+    if uncovered:
+        return StopReport(
+            should_stop=False,
+            reason="coverage_units_missing",
+            candidate_set_hash=candidate_set.candidate_set_hash,
+            uncovered_unit_ids=uncovered,
+            eligible_count=len(candidate_set.eligible),
+            blocked_count=len(candidate_set.blocked),
+        )
+    if candidate_set.eligible:
+        return StopReport(
+            should_stop=False,
+            reason="eligible_candidates_remain",
+            candidate_set_hash=candidate_set.candidate_set_hash,
+            eligible_count=len(candidate_set.eligible),
+            blocked_count=len(candidate_set.blocked),
+        )
+    return StopReport(
+        should_stop=True,
+        reason="coverage_complete_no_eligible_candidates",
+        candidate_set_hash=candidate_set.candidate_set_hash,
+        eligible_count=0,
+        blocked_count=len(candidate_set.blocked),
+    )
+
+
+def _generate_operator_actions(
+    state: Mapping[str, Any],
+    axis: AxisSpec,
+    operator: OperatorSpec,
+) -> tuple[list[CandidateAction], list[BlockedCandidate]]:
+    if axis.kind == "ordered_lattice":
+        return _ordered_lattice_actions(state, axis, operator)
+    return _bounded_numeric_actions(state, axis, operator)
+
+
+def _ordered_lattice_actions(
+    state: Mapping[str, Any],
+    axis: AxisSpec,
+    operator: OperatorSpec,
+) -> tuple[list[CandidateAction], list[BlockedCandidate]]:
+    if operator.kind not in {"bracket", "step_up", "step_down"}:
+        raise ValueError(
+            f"operator {operator.name!r} is not valid for ordered lattice axis {axis.name!r}"
+        )
+    current = state.get(axis.name)
+    if current not in axis.values:
+        raise ValueError(f"state value {current!r} is not in lattice axis {axis.name!r}")
+    index = axis.values.index(current)
+    if operator.kind == "bracket":
+        targets = [value for value in axis.values if value != current]
+        return ([_candidate(axis, operator, current, target) for target in targets], [])
+    if operator.kind == "step_up":
+        if index == len(axis.values) - 1:
+            return (
+                [],
+                [_boundary_block(axis, operator, current, "ordered_lattice_upper_boundary")],
+            )
+        return ([_candidate(axis, operator, current, axis.values[index + 1])], [])
+    if index == 0:
+        return (
+            [],
+            [_boundary_block(axis, operator, current, "ordered_lattice_lower_boundary")],
+        )
+    return ([_candidate(axis, operator, current, axis.values[index - 1])], [])
+
+
+def _bounded_numeric_actions(
+    state: Mapping[str, Any],
+    axis: AxisSpec,
+    operator: OperatorSpec,
+) -> tuple[list[CandidateAction], list[BlockedCandidate]]:
+    if operator.kind not in {"jump_to_floor", "local_climb"}:
+        raise ValueError(
+            f"operator {operator.name!r} is not valid for bounded numeric axis {axis.name!r}"
+        )
+    current = _as_float(state.get(axis.name), axis=axis.name)
+    assert axis.floor is not None
+    assert axis.ceiling is not None
+    assert axis.step is not None
+    if operator.kind == "jump_to_floor":
+        if current < axis.floor:
+            return ([_candidate(axis, operator, current, axis.floor)], [])
+        return ([], [_boundary_block(axis, operator, current, "numeric_at_or_above_floor")])
+    if current < axis.floor:
+        return ([], [_boundary_block(axis, operator, current, "numeric_below_floor")])
+    if current >= axis.ceiling:
+        return ([], [_boundary_block(axis, operator, current, "numeric_upper_boundary")])
+    target = min(axis.ceiling, current + axis.step)
+    return ([_candidate(axis, operator, current, target)], [])
+
+
+def _candidate(axis: AxisSpec, operator: OperatorSpec, source: Any, target: Any) -> CandidateAction:
+    coverage = CoverageUnit(axis=axis.name, operator=operator.kind, target=target)
+    return CandidateAction(
+        action_id=f"{operator.name}:{axis.name}:{_stable_token(source)}->{_stable_token(target)}",
+        operator=operator.name,
+        axis=axis.name,
+        patch={axis.name: target},
+        harness_priority=operator.harness_priority,
+        coverage_units=(coverage,),
+        source_value=source,
+        target_value=target,
+    )
+
+
+def _boundary_block(axis: AxisSpec, operator: OperatorSpec, current: Any, reason: str) -> BlockedCandidate:
+    candidate = CandidateAction(
+        action_id=f"{operator.name}:{axis.name}:{_stable_token(current)}->boundary",
+        operator=operator.name,
+        axis=axis.name,
+        patch={axis.name: current},
+        harness_priority=operator.harness_priority,
+        coverage_units=(),
+        source_value=current,
+        target_value=current,
+    )
+    return BlockedCandidate(candidate=candidate, reason=reason)
+
+
+def _blocking_reason(
+    candidate: CandidateAction,
+    policy: HarnessPolicy,
+    coverage_state: CoverageState,
+) -> str | None:
+    if policy.no_repeat and candidate.signature in coverage_state.tested_signatures:
+        return "no_repeat: signature already tested"
+    for region in coverage_state.failed_regions:
+        if region.contains(candidate):
+            return f"failure_region:{region.axis}:{region.relation}:{_stable_token(region.value)}:{region.reason}"
+    return None
+
+
+def _candidate_set_hash(
+    eligible: tuple[CandidateAction, ...],
+    blocked: tuple[BlockedCandidate, ...],
+) -> str:
+    payload = {
+        "eligible": [_candidate_payload(candidate) for candidate in eligible],
+        "blocked": [
+            {"candidate": _candidate_payload(item.candidate), "reason": item.reason}
+            for item in blocked
+        ],
+    }
+    encoded = json.dumps(
+        payload,
+        sort_keys=True,
+        separators=(",", ":"),
+        ensure_ascii=False,
+    ).encode("utf-8")
+    return hashlib.sha256(encoded).hexdigest()
+
+
+def _candidate_payload(candidate: CandidateAction) -> dict[str, Any]:
+    return {
+        "action_id": candidate.action_id,
+        "axis": candidate.axis,
+        "operator": candidate.operator,
+        "patch": dict(candidate.patch),
+        "harness_priority": candidate.harness_priority,
+        "planner_score": candidate.planner_score,
+        "backend_score": candidate.backend_score,
+        "coverage_unit_ids": [unit.unit_id for unit in candidate.coverage_units],
+        "source_value": candidate.source_value,
+        "target_value": candidate.target_value,
+    }
+
+
+def _candidate_sort_key(candidate: CandidateAction) -> tuple[float, str, str]:
+    return (-candidate.harness_priority, candidate.axis, candidate.action_id)
+
+
+def _stable_token(value: Any) -> str:
+    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
+
+
+def _as_float(value: Any, *, axis: str) -> float:
+    if isinstance(value, bool) or not isinstance(value, (int, float)):
+        raise ValueError(f"state value for numeric axis {axis!r} must be numeric")
+    return float(value)

src/aituner/engine_adapters/__init__.py

@@ -0,0 +1,2 @@
+from __future__ import annotations
+

src/aituner/engine_adapters/vllm.py

@@ -0,0 +1,82 @@
+from __future__ import annotations
+
+from collections.abc import Iterable
+
+from ..knob_descriptor import KnobConstraints, KnobDescriptor
+
+
+def default_vllm_descriptors(*, tunable_flags: Iterable[str]) -> tuple[KnobDescriptor, ...]:
+    tunable = set(tunable_flags)
+    descriptors: list[KnobDescriptor] = []
+    if "max-num-seqs" in tunable:
+        descriptors.append(
+            KnobDescriptor(
+                name="max-num-seqs",
+                location="flag",
+                value_type="int",
+                mechanisms=("admission_capacity", "kv_memory_pressure"),
+                search_geometry="positive_capacity",
+                operators=("coordinate_line_search", "frontier_delta_projection"),
+                constraints=KnobConstraints(min_value=1, integer=True, multiple_of=8),
+                directional_effects={
+                    "increase": ("admission_capacity",),
+                    "decrease": ("kv_memory_pressure",),
+                },
+                risk_effects={
+                    "increase": ("kv_memory_pressure", "decode_tail_latency"),
+                },
+            )
+        )
+    if "max-num-batched-tokens" in tunable:
+        descriptors.append(
+            KnobDescriptor(
+                name="max-num-batched-tokens",
+                location="flag",
+                value_type="int",
+                mechanisms=("prefill_scheduling", "decode_batching"),
+                search_geometry="positive_capacity",
+                operators=("coordinate_line_search", "frontier_delta_projection"),
+                constraints=KnobConstraints(min_value=1, integer=True, multiple_of=128),
+                directional_effects={
+                    "increase": ("prefill_scheduling", "decode_batching"),
+                    "decrease": ("prefill_tail_latency",),
+                },
+                risk_effects={
+                    "increase": ("prefill_tail_latency", "kv_memory_pressure"),
+                },
+            )
+        )
+    if "gpu-memory-utilization" in tunable:
+        descriptors.append(
+            KnobDescriptor(
+                name="gpu-memory-utilization",
+                location="flag",
+                value_type="float",
+                mechanisms=("kv_memory_capacity", "launch_feasibility"),
+                search_geometry="bounded_fraction",
+                operators=("coordinate_line_search", "frontier_delta_projection"),
+                constraints=KnobConstraints(min_value=0.0, max_value=0.97),
+                directional_effects={
+                    "increase": ("kv_memory_capacity",),
+                    "decrease": ("launch_feasibility",),
+                },
+                risk_effects={
+                    "increase": ("launch_feasibility",),
+                },
+            )
+        )
+    if "enable-chunked-prefill" in tunable:
+        descriptors.append(
+            KnobDescriptor(
+                name="enable-chunked-prefill",
+                location="flag",
+                value_type="bool",
+                mechanisms=("prefill_scheduling",),
+                search_geometry="toggle",
+                operators=("coordinate_line_search",),
+                directional_effects={
+                    "toggle": ("prefill_scheduling",),
+                },
+            )
+        )
+    return tuple(descriptors)

src/aituner/interaction_matrix.py

@@ -0,0 +1,424 @@
+from __future__ import annotations
+
+import itertools
+import math
+from typing import Any
+
+from .spec import SpecError, StudySpec
+
+
+AXIS_KNOBS = (
+    "tensor-parallel-size",
+    "max-num-seqs",
+    "max-num-batched-tokens",
+)
+
+
+def build_interaction_screening_matrix(
+    *,
+    study: StudySpec,
+    window_summary: dict[str, Any],
+) -> dict[str, Any]:
+    """Build the low/high screening matrix for knob-interaction experiments.
+
+    This does not launch any run. It only materializes a reviewable set of fixed
+    config patches for a two-level factorial screen over TP, MNS, and MBT.
+    """
+
+    _require_axis_knobs(study)
+    tp_axis = _tp_axis(study)
+    mns_axis = _mns_axis(study, window_summary)
+    mbt_axis = _mbt_axis(window_summary)
+    axes = {
+        "tp": tp_axis,
+        "mns": mns_axis,
+        "mbt": mbt_axis,
+    }
+    configs = _corner_configs(axes)
+    return {
+        "schema_version": 1,
+        "matrix_kind": "low_high_factorial_screening",
+        "study_id": study.study_id,
+        "objective": (
+            "Estimate whether TP, max-num-seqs, and max-num-batched-tokens "
+            "have non-separable effects on feasible request_rate_per_gpu."
+        ),
+        "source": {
+            "window_id": study.trace.window_id,
+            "request_mode": study.trace.request_mode,
+            "engine_name": study.engine.engine_name,
+            "engine_version": study.engine.engine_version,
+            "hardware_gpu_count": study.hardware.gpu_count,
+        },
+        "window_summary": {
+            key: window_summary.get(key)
+            for key in (
+                "request_count",
+                "duration_s",
+                "request_rate",
+                "prompt_tokens_p50",
+                "prompt_tokens_p95",
+                "prompt_tokens_p99",
+                "arrival_qps_1s_p50",
+                "arrival_qps_1s_p95",
+                "arrival_burst_ratio_p95_to_mean",
+            )
+        },
+        "axes": axes,
+        "fixed_tunable_base_flags": _fixed_tunable_base_flags(study, axes),
+        "configs": configs,
+        "recommended_repeats": [
+            {
+                "config_id": configs[0]["config_id"],
+                "reason": "Estimate low-capacity corner noise and baseline drift.",
+            },
+            {
+                "config_id": configs[-1]["config_id"],
+                "reason": "Estimate high-capacity joint-corner noise.",
+            },
+        ],
+        "analysis_plan": {
+            "primary_metric": "max feasible request_rate_per_gpu at target pass_rate",
+            "plot_1": (
+                "2x2 MNS/MBT corner heatmaps faceted by TP low/high, colored by "
+                "request_rate_per_gpu and hatched when infeasible."
+            ),
+            "plot_2": (
+                "Conditional-effect lines: MNS low/high on x-axis, one line per "
+                "TP, faceted by MBT low/high."
+            ),
+            "plot_3": (
+                "Interaction-contrast bars for TPxMNS, TPxMBT, MNSxMBT, and "
+                "TPxMNSxMBT normalized by the low/low/low corner."
+            ),
+        },
+        "review_notes": [
+            "This matrix is for hypothesis testing only; it does not use LLM or harness tuning.",
+            "All non-axis tunable base flags remain fixed and are listed in fixed_tunable_base_flags.",
+            "If a high level is launch-unsafe in a real run, lower it to the largest launch-safe value and keep the blocked corner in the report.",
+        ],
+    }
+
+
+def _require_axis_knobs(study: StudySpec) -> None:
+    tunable = set(study.engine.tunable_flags)
+    missing = [knob for knob in AXIS_KNOBS if knob not in tunable]
+    if missing:
+        raise SpecError(
+            "interaction screening requires tunable flags: "
+            + ", ".join(AXIS_KNOBS)
+            + f". Missing: {', '.join(missing)}."
+        )
+
+
+def _tp_axis(study: StudySpec) -> dict[str, Any]:
+    base_flags = study.engine.base_flags
+    base_tp = _int_flag(base_flags.get("tensor-parallel-size"), default=1)
+    base_dp = _int_flag(base_flags.get("data-parallel-size"), default=1)
+    constraints = study.engine.topology_constraints
+    if constraints is not None and constraints.allowed_tensor_parallel_sizes:
+        raw_values = constraints.allowed_tensor_parallel_sizes
+    else:
+        raw_values = [1, 2, 4, 8]
+    legal = [
+        int(value)
+        for value in sorted(set(raw_values))
+        if _legal_tp_with_fixed_dp(study, tp=int(value), dp=base_dp)
+    ]
+    if len(legal) < 2:
+        if "data-parallel-size" not in set(study.engine.tunable_flags):
+            raise SpecError(
+                "interaction screening needs at least two legal TP levels with fixed "
+                f"data-parallel-size={base_dp}. Legal levels: {legal}."
+            )
+        topologies = _legal_topologies(study)
+        if len(topologies) < 2:
+            raise SpecError(
+                "interaction screening needs at least two legal TP/DP topologies. "
+                f"Legal topologies: {topologies}."
+            )
+        low_topology, high_topology = _adjacent_topology_pair(topologies, base_tp)
+        return {
+            "knob": "topology",
+            "primary_knob": "tensor-parallel-size",
+            "levels": {"low": low_topology, "high": high_topology},
+            "level_labels": {
+                "low": _topology_label(low_topology),
+                "high": _topology_label(high_topology),
+            },
+            "all_legal_topologies": topologies,
+            "scale_rule": (
+                "Fixed-DP TP levels were unavailable, so choose an adjacent legal "
+                "TP/DP redistribution around the base TP."
+            ),
+            "base_value": {
+                "tensor-parallel-size": base_tp,
+                "data-parallel-size": base_dp,
+            },
+        }
+    low_tp, high_tp = _adjacent_tp_pair(legal, base_tp)
+    low = {"tensor-parallel-size": low_tp}
+    high = {"tensor-parallel-size": high_tp}
+    if base_dp != 1:
+        low["data-parallel-size"] = base_dp
+        high["data-parallel-size"] = base_dp
+    return {
+        "knob": "tensor-parallel-size",
+        "levels": {"low": low, "high": high},
+        "level_labels": {"low": _topology_label(low), "high": _topology_label(high)},
+        "all_legal_values_with_fixed_dp": legal,
+        "fixed_data_parallel_size": base_dp,
+        "scale_rule": (
+            "Choose an adjacent legal TP pair around the base TP when possible; "
+            "otherwise use the nearest legal pair. DP is fixed to the base value."
+        ),
+        "base_value": base_tp,
+    }
+
+
+def _mns_axis(study: StudySpec, window_summary: dict[str, Any]) -> dict[str, Any]:
+    arrival_p95 = _float_value(window_summary.get("arrival_qps_1s_p95"))
+    request_rate = _float_value(window_summary.get("request_rate"))
+    concurrency_scale = max(arrival_p95, request_rate, 1.0)
+    low = _round_up_to_multiple(int(math.ceil(0.75 * concurrency_scale)), 8)
+    high = _round_up_to_multiple(int(math.ceil(3.0 * concurrency_scale)), 8)
+    if high <= low:
+        high = low * 2
+    return {
+        "knob": "max-num-seqs",
+        "levels": {"low": low, "high": high},
+        "scale": {
+            "concurrency_scale": concurrency_scale,
+            "arrival_qps_1s_p95": arrival_p95,
+            "request_rate": request_rate,
+            "trace_max_concurrency": study.trace.max_concurrency,
+        },
+        "normalized_levels": {
+            "low": round(low / concurrency_scale, 4),
+            "high": round(high / concurrency_scale, 4),
+        },
+        "scale_rule": (
+            "low=round_up_to_8(0.75*C), high=round_up_to_8(3.0*C), "
+            "where C=max(arrival_qps_1s_p95, request_rate, 1)."
+        ),
+    }
+
+
+def _mbt_axis(window_summary: dict[str, Any]) -> dict[str, Any]:
+    prompt_p95 = _float_value(window_summary.get("prompt_tokens_p95"))
+    prompt_p99 = _float_value(window_summary.get("prompt_tokens_p99"))
+    prompt_scale = prompt_p95 if prompt_p95 > 0 else max(prompt_p99, 1.0)
+    high_cap = 32768
+    low = _round_up_to_multiple(int(math.ceil(prompt_scale)), 1024)
+    high = _round_up_to_multiple(int(math.ceil(4.0 * prompt_scale)), 1024)
+    high = min(high, high_cap)
+    notes: list[str] = []
+    if 0 < high_cap - high <= 1024:
+        high = high_cap
+        notes.append(
+            "high was snapped to 32768 because round_up(4*prompt_scale) was within one 1024-token step of the cap."
+        )
+    if low >= high:
+        low = max(1024, _round_down_to_multiple(high // 2, 1024))
+        notes.append(
+            "low was reduced below prompt_scale because the default high cap would collapse the axis."
+        )
+    return {
+        "knob": "max-num-batched-tokens",
+        "levels": {"low": low, "high": high},
+        "scale": {
+            "prompt_scale": prompt_scale,
+            "prompt_tokens_p95": prompt_p95,
+            "prompt_tokens_p99": prompt_p99,
+            "default_high_cap": high_cap,
+        },
+        "normalized_levels": {
+            "low": round(low / prompt_scale, 4),
+            "high": round(high / prompt_scale, 4),
+        },
+        "scale_rule": (
+            "low=round_up_to_1024(prompt_p95), high=min(round_up_to_1024(4*prompt_p95), 32768)."
+        ),
+        "notes": notes,
+    }
+
+
+def _corner_configs(axes: dict[str, dict[str, Any]]) -> list[dict[str, Any]]:
+    configs: list[dict[str, Any]] = []
+    for tp_level, mns_level, mbt_level in itertools.product(("low", "high"), repeat=3):
+        topology_patch = dict(axes["tp"]["levels"][tp_level])
+        mns = axes["mns"]["levels"][mns_level]
+        mbt = axes["mbt"]["levels"][mbt_level]
+        config_id = f"tp-{tp_level}_mns-{mns_level}_mbt-{mbt_level}"
+        flag_patch = {
+            **topology_patch,
+            "max-num-seqs": mns,
+            "max-num-batched-tokens": mbt,
+        }
+        configs.append(
+            {
+                "config_id": config_id,
+                "levels": {
+                    "tp": tp_level,
+                    "mns": mns_level,
+                    "mbt": mbt_level,
+                },
+                "config_patch": {
+                    "env_patch": {},
+                    "flag_patch": flag_patch,
+                },
+                "normalized_coordinates": {
+                    "topology": axes["tp"]["level_labels"][tp_level],
+                    "mns_over_concurrency_scale": round(
+                        mns / axes["mns"]["scale"]["concurrency_scale"], 4
+                    ),
+                    "mbt_over_prompt_scale": round(
+                        mbt / axes["mbt"]["scale"]["prompt_scale"], 4
+                    ),
+                },
+            }
+        )
+    return configs
+
+
+def _fixed_tunable_base_flags(study: StudySpec, axes: dict[str, dict[str, Any]]) -> dict[str, Any]:
+    tunable = set(study.engine.tunable_flags)
+    axis = set(AXIS_KNOBS)
+    tp_levels = axes.get("tp", {}).get("levels")
+    if isinstance(tp_levels, dict):
+        for level in tp_levels.values():
+            if isinstance(level, dict):
+                axis.update(str(key) for key in level)
+    return {
+        key: value
+        for key, value in study.engine.base_flags.items()
+        if key in tunable and key not in axis
+    }
+
+
+def _legal_tp_with_fixed_dp(study: StudySpec, *, tp: int, dp: int) -> bool:
+    return _legal_topology(study, tp=tp, dp=dp)
+
+
+def _legal_topologies(study: StudySpec) -> list[dict[str, int]]:
+    constraints = study.engine.topology_constraints
+    if constraints is not None and constraints.allowed_tensor_parallel_sizes:
+        tp_values = constraints.allowed_tensor_parallel_sizes
+    else:
+        tp_values = [1, 2, 4, 8]
+    if constraints is not None and constraints.allowed_data_parallel_sizes:
+        dp_values = constraints.allowed_data_parallel_sizes
+    else:
+        dp_values = [1]
+    topologies = []
+    for tp in sorted(set(int(item) for item in tp_values)):
+        for dp in sorted(set(int(item) for item in dp_values)):
+            if _legal_topology(study, tp=tp, dp=dp):
+                topologies.append(
+                    {
+                        "tensor-parallel-size": tp,
+                        "data-parallel-size": dp,
+                    }
+                )
+    topologies.sort(key=lambda item: (item["tensor-parallel-size"], item["data-parallel-size"]))
+    return topologies
+
+
+def _legal_topology(study: StudySpec, *, tp: int, dp: int) -> bool:
+    constraints = study.engine.topology_constraints
+    product = tp * dp
+    if constraints is None:
+        return product <= study.hardware.gpu_count
+    if constraints.allowed_tensor_parallel_sizes and tp not in constraints.allowed_tensor_parallel_sizes:
+        return False
+    if constraints.allowed_data_parallel_sizes and dp not in constraints.allowed_data_parallel_sizes:
+        return False
+    if constraints.allowed_tp_dp_products and product not in constraints.allowed_tp_dp_products:
+        return False
+    if not constraints.allowed_tp_dp_products and product > study.hardware.gpu_count:
+        return False
+    if constraints.require_tp_dp_product_equals_gpu_count and product != study.hardware.gpu_count:
+        return False
+    return True
+
+
+def _adjacent_tp_pair(values: list[int], base_tp: int) -> tuple[int, int]:
+    if base_tp in values:
+        idx = values.index(base_tp)
+        if idx < len(values) - 1:
+            return values[idx], values[idx + 1]
+        return values[idx - 1], values[idx]
+    larger = [value for value in values if value > base_tp]
+    if larger:
+        high = larger[0]
+        idx = values.index(high)
+        if idx > 0:
+            return values[idx - 1], high
+        return high, values[idx + 1]
+    return values[-2], values[-1]
+
+
+def _adjacent_topology_pair(
+    topologies: list[dict[str, int]],
+    base_tp: int,
+) -> tuple[dict[str, int], dict[str, int]]:
+    by_tp = []
+    seen: set[int] = set()
+    for item in topologies:
+        tp = item["tensor-parallel-size"]
+        if tp in seen:
+            continue
+        seen.add(tp)
+        by_tp.append(item)
+    if len(by_tp) < 2:
+        return topologies[0], topologies[-1]
+    tp_values = [item["tensor-parallel-size"] for item in by_tp]
+    if base_tp in tp_values:
+        idx = tp_values.index(base_tp)
+        if idx < len(by_tp) - 1:
+            return by_tp[idx], by_tp[idx + 1]
+        return by_tp[idx - 1], by_tp[idx]
+    larger = [idx for idx, tp in enumerate(tp_values) if tp > base_tp]
+    if larger:
+        idx = larger[0]
+        if idx > 0:
+            return by_tp[idx - 1], by_tp[idx]
+        return by_tp[idx], by_tp[idx + 1]
+    return by_tp[-2], by_tp[-1]
+
+
+def _topology_label(topology: dict[str, int]) -> str:
+    tp = topology.get("tensor-parallel-size")
+    dp = topology.get("data-parallel-size", 1)
+    return f"TP{tp}/DP{dp}"
+
+
+def _int_flag(value: Any, *, default: int) -> int:
+    if isinstance(value, bool):
+        return default
+    if isinstance(value, int):
+        return value
+    if isinstance(value, float) and value.is_integer():
+        return int(value)
+    if isinstance(value, str) and value.strip():
+        try:
+            return int(value)
+        except ValueError:
+            return default
+    return default
+
+
+def _float_value(value: Any) -> float:
+    if isinstance(value, bool):
+        return 0.0
+    if isinstance(value, (int, float)):
+        return max(0.0, float(value))
+    return 0.0
+
+
+def _round_up_to_multiple(value: int, multiple: int) -> int:
+    return ((max(value, 1) + multiple - 1) // multiple) * multiple
+
+
+def _round_down_to_multiple(value: int, multiple: int) -> int:
+    return max(multiple, (max(value, 1) // multiple) * multiple)

src/aituner/knob_descriptor.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Literal, Mapping
+
+
+KnobLocation = Literal["flag", "env"]
+KnobValueType = Literal["int", "float", "bool", "enum", "str"]
+SearchGeometry = Literal["linear", "positive_capacity", "bounded_fraction", "toggle"]
+
+
+@dataclass(frozen=True)
+class KnobConstraints:
+    min_value: float | None = None
+    max_value: float | None = None
+    integer: bool = False
+    multiple_of: int | None = None
+
+
+@dataclass(frozen=True)
+class KnobDescriptor:
+    """Declarative serving-knob semantics used by generic planners.
+
+    The descriptor intentionally does not enumerate target values for continuous
+    or large integer knobs. It describes how a generic operator may perturb the
+    knob and which mechanism each direction is expected to affect.
+    """
+
+    name: str
+    location: KnobLocation
+    value_type: KnobValueType
+    mechanisms: tuple[str, ...]
+    search_geometry: SearchGeometry
+    operators: tuple[str, ...]
+    constraints: KnobConstraints = field(default_factory=KnobConstraints)
+    directional_effects: Mapping[str, tuple[str, ...]] = field(default_factory=dict)
+    risk_effects: Mapping[str, tuple[str, ...]] = field(default_factory=dict)
+
+    def current_value(self, config: Mapping[str, Any]) -> Any:
+        return config.get(self.name)

src/aituner/llm.py

@@ -5,9 +5,10 @@ import time
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
-from .harness import build_harness_context, render_harness_context
+from .engine_adapters.vllm import default_vllm_descriptors
+from .harness import _effective_config_signature, build_harness_context, render_harness_context
 from .http_client import chat_completion, stream_text_completion
-from .spec import LLMPolicySpec, Proposal, SpecError, StudySpec, StudyState
+from .spec import LLMPolicySpec, Proposal, SpecError, StudySpec, StudyState, to_jsonable
 
 if TYPE_CHECKING:
     from .lca import WorkloadProfile
@@ -175,6 +176,108 @@ def _enumerate_parallel_candidates(study: StudySpec) -> list[dict[str, int | boo
     return candidates
 
 
+def _descriptor_payload(study: StudySpec) -> list[dict[str, Any]]:
+    if study.engine.engine_name.lower() != "vllm":
+        return []
+    return [
+        to_jsonable(descriptor)
+        for descriptor in default_vllm_descriptors(tunable_flags=study.engine.tunable_flags)
+    ]
+
+
+def build_initial_config_review_prompt(
+    *,
+    study: StudySpec,
+    window_summary: dict[str, Any],
+    capability_profile: dict[str, Any] | None,
+    workload_profile: "WorkloadProfile | None" = None,
+) -> str:
+    """Build the static pre-flight audit prompt for the study's base config."""
+
+    payload: dict[str, Any] = {
+        "study_id": study.study_id,
+        "objective": "static initial-config audit before any measurement or repair",
+        "review_mode": study.llm.initial_config_review.mode,
+        "output_contract": {
+            "verdict": "one of ok, risky, invalid, unknown",
+            "issues": [
+                {
+                    "knob": "flag/env knob name or mechanism name",
+                    "mechanism": "affected mechanism if known",
+                    "reason": "short reason grounded in the supplied descriptors/context",
+                    "severity": "low|medium|high",
+                }
+            ],
+            "minimal_repair_patch": {
+                "env_patch": {},
+                "flag_patch": {},
+            },
+            "do_not_change": ["knob names that should remain fixed during repair"],
+            "confidence": "number in [0, 1]",
+            "requires_harness_validation": True,
+        },
+        "hardware": {
+            "gpu_count": study.hardware.gpu_count,
+            "gpu_model": study.hardware.gpu_model,
+        },
+        "model": {
+            "model_id": study.model.model_id,
+            "served_model_name": study.model.served_model_name,
+        },
+        "engine": {
+            "engine_name": study.engine.engine_name,
+            "engine_version": study.engine.engine_version,
+            "base_envs": study.engine.base_envs,
+            "base_flags": study.engine.base_flags,
+            "allowed_env_keys": study.engine.tunable_envs,
+            "allowed_flag_keys": study.engine.tunable_flags,
+            "topology_constraints": (
+                study.engine.topology_constraints.__dict__
+                if study.engine.topology_constraints is not None
+                else None
+            ),
+            "effective_topology": _effective_topology(study),
+        },
+        "trace": {
+            "window_id": study.trace.window_id,
+            "request_mode": study.trace.request_mode,
+            "completion_tokens_override": study.trace.completion_tokens_override,
+            "input_length_filter": (
+                {
+                    "min_input_tokens": study.trace.input_length_filter.min_input_tokens,
+                    "max_input_tokens": study.trace.input_length_filter.max_input_tokens,
+                }
+                if study.trace.input_length_filter is not None
+                else None
+            ),
+            "window_summary": window_summary,
+            "workload_lca_profile": (
+                workload_profile.to_dict() if workload_profile is not None else None
+            ),
+        },
+        "slo": {
+            "target_pass_rate": study.slo.target_pass_rate,
+            "ttft_rule": study.slo.ttft_rule,
+            "tpot_rule": study.slo.tpot_rule,
+        },
+        "capability_profile": capability_profile or {},
+        "knob_descriptors": _descriptor_payload(study),
+    }
+    sections = [
+        "You are doing a pre-flight review of an LLM serving initial configuration.",
+        "This is not a tuning proposal. Do not claim measured performance.",
+        "Identify obviously risky or inconsistent initial settings before the first baseline trial.",
+        "Use the supplied knob descriptors and constraints; do not use case-specific memorized values.",
+        "Return exactly one JSON object matching output_contract. Do not wrap it in markdown.",
+        "If uncertain, use verdict=unknown and an empty minimal_repair_patch.",
+        "The current warn mode records your audit only; the patch will not be applied automatically.",
+        "",
+        "Audit context:",
+        json.dumps(payload, default=lambda value: value.__dict__, ensure_ascii=False, indent=2),
+    ]
+    return "\n".join(sections)
+
+
 def build_prompt(
     *,
     study: StudySpec,
@@ -306,7 +409,7 @@ def build_prompt(
             json.dumps(launch_failures, ensure_ascii=False, indent=2),
             "",
             "Tested config signatures:",
-            json.dumps(_tested_config_signatures(state), ensure_ascii=False, indent=2),
+            json.dumps(_tested_config_signatures(study, state), ensure_ascii=False, indent=2),
         ]
         return "\n".join(sections)
 
@@ -317,6 +420,11 @@ def build_prompt(
             if parallel_candidates
             else "If TP/DP/EP are not tunable, focus on the remaining launch-safe runtime knobs."
         ),
+        (
+            "Harness candidate policy is advisory: prefer a high-scoring harness candidate when it matches your diagnosis, but you may propose an out-of-set config when the harness candidate family appears to miss the right step; such proposals are audited as candidate-family gaps."
+            if study.llm.harness_candidate_policy == "advisory"
+            else "Harness candidate policy is strict: your config_patch must match one of the harness eligible candidates after effective-config materialization."
+        ),
         "",
         "Study stack:",
         json.dumps(
@@ -402,7 +510,7 @@ def build_prompt(
         json.dumps(parallel_candidates, ensure_ascii=False, indent=2),
         "",
         "Tested config signatures:",
-        json.dumps(_tested_config_signatures(state), ensure_ascii=False, indent=2),
+        json.dumps(_tested_config_signatures(study, state), ensure_ascii=False, indent=2),
     ]
     sections.extend(
         [
@@ -435,12 +543,12 @@ def build_prompt(
     return "\n".join(sections)
 
 
-def _tested_config_signatures(state: StudyState) -> list[dict[str, Any]]:
+def _tested_config_signatures(study: StudySpec, state: StudyState) -> list[dict[str, Any]]:
     signatures: list[dict[str, Any]] = []
     seen: set[str] = set()
     for trial in state.trials:
         config_patch = trial.config_patch or {}
-        signature = json.dumps(config_patch, sort_keys=True, ensure_ascii=False)
+        signature = _effective_config_signature(study, config_patch)
         if signature in seen:
             continue
         seen.add(signature)
@@ -449,6 +557,7 @@ def _tested_config_signatures(state: StudyState) -> list[dict[str, Any]]:
                 "trial_id": trial.trial_id,
                 "status": trial.status,
                 "best_request_rate_per_gpu": trial.best_request_rate_per_gpu,
+                "effective_config_signature": signature,
                 "config_patch": config_patch,
             }
         )
@@ -638,6 +747,87 @@ def parse_proposal_text(text: str, study: StudySpec) -> Proposal:
     return validate_proposal(proposal, study)
 
 
+def parse_initial_config_review_text(text: str, study: StudySpec) -> dict[str, Any]:
+    payload = _parse_json_object_text(text)
+    verdict = str(payload.get("verdict") or "unknown").strip().lower()
+    if verdict not in {"ok", "risky", "invalid", "unknown"}:
+        raise SpecError(
+            "initial-config review verdict must be one of: ok, risky, invalid, unknown."
+        )
+
+    issues_payload = payload.get("issues", [])
+    if not isinstance(issues_payload, list):
+        raise SpecError("initial-config review issues must be a list.")
+    issues: list[dict[str, str]] = []
+    for idx, item in enumerate(issues_payload):
+        if isinstance(item, str):
+            issues.append(
+                {
+                    "knob": "",
+                    "mechanism": "",
+                    "reason": item.strip(),
+                    "severity": "medium",
+                }
+            )
+            continue
+        if not isinstance(item, dict):
+            raise SpecError(f"initial-config review issues[{idx}] must be an object.")
+        severity = str(item.get("severity") or "medium").strip().lower()
+        if severity not in {"low", "medium", "high"}:
+            severity = "medium"
+        issues.append(
+            {
+                "knob": str(item.get("knob") or "").strip(),
+                "mechanism": str(item.get("mechanism") or "").strip(),
+                "reason": str(item.get("reason") or "").strip(),
+                "severity": severity,
+            }
+        )
+
+    repair_payload = payload.get("minimal_repair_patch") or {}
+    if not isinstance(repair_payload, dict):
+        raise SpecError("initial-config review minimal_repair_patch must be an object.")
+    if "env_patch" not in repair_payload and "flag_patch" not in repair_payload:
+        repair_payload = {"env_patch": {}, "flag_patch": repair_payload}
+    repair_proposal = Proposal.from_dict(
+        {
+            "observation": "Initial-config pre-flight repair candidate.",
+            "diagnosis": "Validate the LLM audit's minimal repair patch against study constraints.",
+            "config_patch": repair_payload,
+            "expected_effects": ["pre-flight audit only"],
+            "should_stop": False,
+        }
+    )
+    validate_proposal(repair_proposal, study)
+
+    do_not_change_payload = payload.get("do_not_change", [])
+    if isinstance(do_not_change_payload, str):
+        do_not_change = [do_not_change_payload.strip()] if do_not_change_payload.strip() else []
+    elif isinstance(do_not_change_payload, list):
+        do_not_change = [str(item).strip() for item in do_not_change_payload if str(item).strip()]
+    else:
+        raise SpecError("initial-config review do_not_change must be a list.")
+
+    raw_confidence = payload.get("confidence", 0.0)
+    confidence = float(raw_confidence) if isinstance(raw_confidence, (int, float)) else 0.0
+    confidence = max(0.0, min(1.0, confidence))
+    requires_validation = payload.get("requires_harness_validation")
+    if requires_validation is None:
+        requires_validation = True
+    if not isinstance(requires_validation, bool):
+        raise SpecError("initial-config review requires_harness_validation must be boolean.")
+
+    return {
+        "schema_version": 1,
+        "verdict": verdict,
+        "issues": issues,
+        "minimal_repair_patch": to_jsonable(repair_proposal.config_patch),
+        "do_not_change": do_not_change,
+        "confidence": confidence,
+        "requires_harness_validation": requires_validation,
+    }
+
+
 def _extract_response_text(response: dict[str, Any]) -> str:
     output_text = response.get("output_text")
     if isinstance(output_text, str) and output_text:
@@ -677,11 +867,11 @@ def _extract_response_text(response: dict[str, Any]) -> str:
     raise RuntimeError("LLM response content is empty")
 
 
-def call_llm_for_proposal(
+def _call_llm_text(
     *,
     policy: LLMPolicySpec,
     prompt: str,
-    use_harness: bool = True,
+    system_prompt: str = "",
 ) -> str:
     if policy.endpoint is None:
         raise RuntimeError("study.llm.endpoint is not configured")
@@ -689,7 +879,6 @@ def call_llm_for_proposal(
     max_attempts = 4
     for attempt in range(max_attempts):
         try:
-            system_prompt = policy.system_prompt if use_harness else ""
             if policy.endpoint.stream:
                 text = stream_text_completion(
                     base_url=policy.endpoint.base_url,
@@ -724,3 +913,29 @@ def call_llm_for_proposal(
             time.sleep(min(30.0, 2.0 * (2**attempt)))
             continue
     raise RuntimeError(f"LLM proposal failed after retry: {last_error}") from last_error
+
+
+def call_llm_for_proposal(
+    *,
+    policy: LLMPolicySpec,
+    prompt: str,
+    use_harness: bool = True,
+) -> str:
+    system_prompt = policy.system_prompt if use_harness else ""
+    return _call_llm_text(policy=policy, prompt=prompt, system_prompt=system_prompt)
+
+
+def call_llm_for_initial_config_review(
+    *,
+    policy: LLMPolicySpec,
+    prompt: str,
+) -> str:
+    review_system = "\n".join(
+        item
+        for item in (
+            policy.system_prompt,
+            "You are auditing an initial serving config. Return only the requested JSON audit.",
+        )
+        if item.strip()
+    )
+    return _call_llm_text(policy=policy, prompt=prompt, system_prompt=review_system)

src/aituner/mechanism_planner.py

@@ -0,0 +1,230 @@
+from __future__ import annotations
+
+import math
+from dataclasses import dataclass, field
+from typing import Any, Mapping
+
+from .knob_descriptor import KnobDescriptor
+
+
+@dataclass(frozen=True)
+class CoordinateSearchPolicy:
+    initial_relative_step: float = 1.0
+    initial_fraction_step: float = 0.05
+    step_multipliers: tuple[float, ...] = (1.0,)
+    grow_factor: float = 1.5
+    shrink_factor: float = 0.5
+    min_score: float = 0.0
+
+
+@dataclass(frozen=True)
+class CoordinateOperatorState:
+    knob: str
+    direction: str
+    trust_radius: float | None = None
+    last_good_value: Any | None = None
+    last_bad_value: Any | None = None
+    tested_values: tuple[Any, ...] = ()
+
+
+@dataclass(frozen=True)
+class MechanismCandidate:
+    action_id: str
+    knob: str
+    mechanism: str
+    operator: str
+    direction: str
+    patch: dict[str, Any]
+    score: float
+    score_factors: dict[str, float]
+    evidence_refs: tuple[str, ...] = ()
+
+
+def coordinate_line_search_candidates(
+    *,
+    current_config: Mapping[str, Any],
+    descriptors: tuple[KnobDescriptor, ...],
+    evidence_weights: Mapping[str, float],
+    states: Mapping[tuple[str, str], CoordinateOperatorState] | None = None,
+    policy: CoordinateSearchPolicy | None = None,
+) -> tuple[MechanismCandidate, ...]:
+    policy = policy or CoordinateSearchPolicy()
+    states = states or {}
+    candidates: list[MechanismCandidate] = []
+    for descriptor in descriptors:
+        if "coordinate_line_search" not in descriptor.operators:
+            continue
+        direction, mechanism, evidence = _choose_direction(descriptor, evidence_weights)
+        if direction is None or mechanism is None:
+            continue
+        if evidence < policy.min_score:
+            continue
+        state = states.get((descriptor.name, direction))
+        current = descriptor.current_value(current_config)
+        multipliers = (
+            (1.0,) if descriptor.search_geometry == "toggle" else policy.step_multipliers
+        )
+        for multiplier in multipliers:
+            target = _propose_value(
+                descriptor=descriptor,
+                current=current,
+                direction=direction,
+                state=state,
+                policy=policy,
+                step_multiplier=multiplier,
+            )
+            if target is None or target == current:
+                continue
+            risk = _direction_risk(descriptor, direction, evidence_weights)
+            step_risk = _step_risk(multiplier)
+            score = max(0.0, evidence - risk - step_risk)
+            candidates.append(
+                MechanismCandidate(
+                    action_id=(
+                        f"coordinate_line_search:{descriptor.search_geometry}:"
+                        f"{descriptor.name}:{direction}:step={_stable_token(multiplier)}:"
+                        f"{_stable_token(current)}->{_stable_token(target)}"
+                    ),
+                    knob=descriptor.name,
+                    mechanism=mechanism,
+                    operator="coordinate_line_search",
+                    direction=direction,
+                    patch={descriptor.name: target},
+                    score=round(score, 4),
+                    score_factors={
+                        "mechanism_evidence": round(evidence, 4),
+                        "direction_risk": round(risk, 4),
+                        "step_risk": round(step_risk, 4),
+                    },
+                    evidence_refs=(mechanism,),
+                )
+            )
+    candidates.sort(key=lambda item: (item.score, item.action_id), reverse=True)
+    return tuple(candidates)
+
+
+def _choose_direction(
+    descriptor: KnobDescriptor,
+    evidence_weights: Mapping[str, float],
+) -> tuple[str | None, str | None, float]:
+    best_direction: str | None = None
+    best_mechanism: str | None = None
+    best_weight = 0.0
+    for direction, mechanisms in descriptor.directional_effects.items():
+        for mechanism in mechanisms:
+            weight = float(evidence_weights.get(mechanism, 0.0))
+            if weight > best_weight:
+                best_direction = direction
+                best_mechanism = mechanism
+                best_weight = weight
+    return best_direction, best_mechanism, best_weight
+
+
+def _direction_risk(
+    descriptor: KnobDescriptor,
+    direction: str,
+    evidence_weights: Mapping[str, float],
+) -> float:
+    risks = descriptor.risk_effects.get(direction, ())
+    if not risks:
+        return 0.0
+    return min(0.5, 0.2 * max(float(evidence_weights.get(item, 0.0)) for item in risks))
+
+
+def _step_risk(multiplier: float) -> float:
+    return min(0.2, max(0.0, float(multiplier) - 1.0) * 0.04)
+
+
+def _propose_value(
+    *,
+    descriptor: KnobDescriptor,
+    current: Any,
+    direction: str,
+    state: CoordinateOperatorState | None,
+    policy: CoordinateSearchPolicy,
+    step_multiplier: float,
+) -> Any | None:
+    if descriptor.search_geometry == "toggle":
+        if not isinstance(current, bool):
+            current = bool(current)
+        return not current
+    if descriptor.search_geometry == "bounded_fraction":
+        value = _as_float(current)
+        if value is None:
+            return None
+        radius = (
+            state.trust_radius
+            if state is not None and state.trust_radius is not None
+            else policy.initial_fraction_step
+        )
+        radius *= max(float(step_multiplier), 0.0)
+        if direction == "decrease":
+            target = value - radius
+        else:
+            target = value + radius
+        return _canonicalize_value(descriptor, target)
+    if descriptor.search_geometry == "linear":
+        value = _as_float(current)
+        if value is None:
+            return None
+        radius = (
+            state.trust_radius
+            if state is not None and state.trust_radius is not None
+            else 1.0
+        )
+        radius *= max(float(step_multiplier), 0.0)
+        return _canonicalize_value(
+            descriptor,
+            value - radius if direction == "decrease" else value + radius,
+        )
+    if descriptor.search_geometry == "positive_capacity":
+        value = _as_float(current)
+        if value is None or value <= 0:
+            min_value = descriptor.constraints.min_value
+            return _canonicalize_value(descriptor, min_value if min_value is not None else 1)
+        radius = (
+            state.trust_radius
+            if state is not None and state.trust_radius is not None
+            else policy.initial_relative_step
+        )
+        radius *= max(float(step_multiplier), 0.0)
+        factor = max(1.0 + radius, 1.01)
+        target = value / factor if direction == "decrease" else value * factor
+        return _canonicalize_value(descriptor, target)
+    raise ValueError(f"unsupported search geometry {descriptor.search_geometry!r}")
+
+
+def _canonicalize_value(descriptor: KnobDescriptor, value: float | int) -> Any:
+    target = float(value)
+    if descriptor.constraints.min_value is not None:
+        target = max(target, float(descriptor.constraints.min_value))
+    if descriptor.constraints.max_value is not None:
+        target = min(target, float(descriptor.constraints.max_value))
+    if descriptor.constraints.integer or descriptor.value_type == "int":
+        integer_target = int(math.ceil(target))
+        multiple_of = descriptor.constraints.multiple_of
+        if multiple_of is not None and multiple_of > 1:
+            integer_target = int(math.ceil(integer_target / multiple_of) * multiple_of)
+        if descriptor.constraints.min_value is not None:
+            integer_target = max(integer_target, int(descriptor.constraints.min_value))
+        if descriptor.constraints.max_value is not None:
+            integer_target = min(integer_target, int(descriptor.constraints.max_value))
+        return integer_target
+    return round(target, 6)
+
+
+def _as_float(value: Any) -> float | None:
+    if isinstance(value, bool):
+        return None
+    if isinstance(value, (int, float)):
+        return float(value)
+    if isinstance(value, str) and value.strip():
+        try:
+            return float(value.strip())
+        except ValueError:
+            return None
+    return None
+
+
+def _stable_token(value: Any) -> str:
+    return repr(value)

src/aituner/spec.py

@@ -585,6 +585,42 @@ class SloSpec:
         )
 
 
+@dataclass(frozen=True)
+class SearchAutoHighSpec:
+    enabled: bool = False
+    max_sampling_u: float = 1.0
+    require_human_confirmation_beyond_trace: bool = True
+
+    @classmethod
+    def from_dict(cls, data: Any) -> "SearchAutoHighSpec":
+        if data is None:
+            return cls()
+        m = _require_mapping(data, context="search.auto_high")
+        enabled = (
+            _require_bool(m.get("enabled"), context="search.auto_high.enabled")
+            if m.get("enabled") is not None
+            else False
+        )
+        max_sampling_u = _require_float(
+            m.get("max_sampling_u", 1.0), context="search.auto_high.max_sampling_u"
+        )
+        if not 0.0 < max_sampling_u <= 1.0:
+            raise SpecError("search.auto_high.max_sampling_u must be in (0, 1].")
+        require_confirmation = (
+            _require_bool(
+                m.get("require_human_confirmation_beyond_trace"),
+                context="search.auto_high.require_human_confirmation_beyond_trace",
+            )
+            if m.get("require_human_confirmation_beyond_trace") is not None
+            else True
+        )
+        return cls(
+            enabled=enabled,
+            max_sampling_u=max_sampling_u,
+            require_human_confirmation_beyond_trace=require_confirmation,
+        )
+
+
 @dataclass(frozen=True)
 class SamplingSearchSpec:
     low: float
@@ -593,16 +629,27 @@ class SamplingSearchSpec:
     max_probes: int
     sample_seed: int
     inherit_incumbent_floor: bool = False
+    auto_high: SearchAutoHighSpec = field(default_factory=SearchAutoHighSpec)
 
     @classmethod
     def from_dict(cls, data: Mapping[str, Any]) -> "SamplingSearchSpec":
+        low = _require_float(data.get("low", 0.0), context="search.low")
+        high = _require_float(data.get("high", 1.0), context="search.high")
+        tolerance = _require_float(data.get("tolerance", 0.01), context="search.tolerance")
+        max_probes = _require_int(data.get("max_probes", 8), context="search.max_probes")
+        if low < 0:
+            raise SpecError("search.low must be >= 0.")
+        if high < low:
+            raise SpecError("search.high must be >= search.low.")
+        if tolerance <= 0:
+            raise SpecError("search.tolerance must be > 0.")
+        if max_probes <= 0:
+            raise SpecError("search.max_probes must be > 0.")
         return cls(
-            low=_require_float(data.get("low", 0.0), context="search.low"),
+            low=low,
-            high=_require_float(data.get("high", 1.0), context="search.high"),
+            high=high,
-            tolerance=_require_float(
+            tolerance=tolerance,
-                data.get("tolerance", 0.01), context="search.tolerance"
+            max_probes=max_probes,
-            ),
-            max_probes=_require_int(data.get("max_probes", 8), context="search.max_probes"),
             sample_seed=_require_int(
                 data.get("sample_seed", 20260325), context="search.sample_seed"
             ),
@@ -610,6 +657,7 @@ class SamplingSearchSpec:
                 data.get("inherit_incumbent_floor", False),
                 context="search.inherit_incumbent_floor",
             ),
+            auto_high=SearchAutoHighSpec.from_dict(data.get("auto_high")),
         )
 
 
@@ -678,12 +726,31 @@ class LLMEndpointSpec:
         )
 
 
+@dataclass(frozen=True)
+class InitialConfigReviewSpec:
+    mode: str = "off"
+
+    @classmethod
+    def from_dict(cls, data: Any) -> "InitialConfigReviewSpec":
+        if data is None:
+            return cls()
+        payload = _require_mapping(data, context="llm.initial_config_review")
+        mode = str(payload.get("mode") or "off").strip().lower()
+        if mode not in {"off", "warn"}:
+            raise SpecError("llm.initial_config_review.mode must be one of: off, warn.")
+        return cls(mode=mode)
+
+
 @dataclass(frozen=True)
 class LLMPolicySpec:
     endpoint: LLMEndpointSpec | None
     system_prompt: str
     max_history_trials: int
     use_harness: bool = True
+    harness_candidate_policy: str = "advisory"
+    initial_config_review: InitialConfigReviewSpec = field(
+        default_factory=InitialConfigReviewSpec
+    )
 
     @classmethod
     def from_dict(cls, data: Mapping[str, Any] | None) -> "LLMPolicySpec":
@@ -695,6 +762,13 @@ class LLMPolicySpec:
             if payload.get("endpoint")
             else None
         )
+        harness_candidate_policy = str(
+            payload.get("harness_candidate_policy") or "advisory"
+        ).strip()
+        if harness_candidate_policy not in {"advisory", "strict"}:
+            raise SpecError(
+                "llm.harness_candidate_policy must be one of: advisory, strict."
+            )
         return cls(
             endpoint=endpoint,
             system_prompt=str(payload.get("system_prompt") or "").strip(),
@@ -706,6 +780,10 @@ class LLMPolicySpec:
                 if payload.get("use_harness") is not None
                 else True
             ),
+            harness_candidate_policy=harness_candidate_policy,
+            initial_config_review=InitialConfigReviewSpec.from_dict(
+                payload.get("initial_config_review")
+            ),
         )
 
 
@@ -823,6 +901,7 @@ class TrialSpec:
     probe_log_path: str
     engine_log_path: str
     result_path: str
+    search_evidence: dict[str, Any] = field(default_factory=dict)
 
 
 @dataclass

src/aituner/store.py

@@ -5,15 +5,16 @@ from dataclasses import replace
 from pathlib import Path
 from typing import Any
 
-from .spec import ConfigPatch, Proposal, StudySpec, StudyState, TrialSpec, TrialSummary, to_jsonable
+from .config_signature import materialize_proposal_for_execution
-
+from .spec import (
-
+    Proposal,
-_TOPOLOGY_FLAG_KEYS = {
+    SamplingSearchSpec,
-    "tensor-parallel-size",
+    StudySpec,
-    "data-parallel-size",
+    StudyState,
-    "expert-parallel-size",
+    TrialSpec,
-    "enable-expert-parallel",
+    TrialSummary,
-}
+    to_jsonable,
+)
 
 
 class StudyStore:
@@ -26,7 +27,16 @@ class StudyStore:
 
     def init_study(self, *, spec_path: Path, study: StudySpec) -> Path:
         root = self.study_root(study.study_id)
-        for rel in ("prompts", "proposals", "trials", "results"):
+        for rel in (
+            "prompts",
+            "proposals",
+            "proposal_attributions",
+            "trials",
+            "results",
+            "harness",
+            "candidate_family_gaps",
+            "preflight_audits",
+        ):
             (root / rel).mkdir(parents=True, exist_ok=True)
         (root / "study_spec.source").write_text(str(spec_path.resolve()) + "\n", encoding="utf-8")
         self.write_json(root / "study_spec.snapshot.json", to_jsonable(study))
@@ -69,6 +79,46 @@ class StudyStore:
         self.write_json(path, to_jsonable(proposal))
         return path
 
+    def write_harness_snapshot(
+        self,
+        study_id: str,
+        snapshot_name: str,
+        payload: dict[str, Any],
+    ) -> Path:
+        path = self.study_root(study_id) / "harness" / f"{snapshot_name}.json"
+        self.write_json(path, payload)
+        return path
+
+    def write_proposal_attribution(
+        self,
+        study_id: str,
+        proposal_name: str,
+        payload: dict[str, Any],
+    ) -> Path:
+        path = self.study_root(study_id) / "proposal_attributions" / f"{proposal_name}.json"
+        self.write_json(path, payload)
+        return path
+
+    def write_candidate_family_gap(
+        self,
+        study_id: str,
+        trial_id: str,
+        payload: dict[str, Any],
+    ) -> Path:
+        path = self.study_root(study_id) / "candidate_family_gaps" / f"{trial_id}.json"
+        self.write_json(path, payload)
+        return path
+
+    def write_preflight_audit(
+        self,
+        study_id: str,
+        audit_name: str,
+        payload: dict[str, Any],
+    ) -> Path:
+        path = self.study_root(study_id) / "preflight_audits" / f"{audit_name}.json"
+        self.write_json(path, payload)
+        return path
+
     def materialize_trial(
         self,
         *,
@@ -76,7 +126,7 @@ class StudyStore:
         state: StudyState,
         proposal: Proposal,
     ) -> tuple[TrialSpec, StudyState]:
-        proposal = _inherit_incumbent_topology_for_runtime_patch(
+        proposal = materialize_proposal_for_execution(
             study=study,
             state=state,
             proposal=proposal,
@@ -95,6 +145,13 @@ class StudyStore:
                     parallel_size=parallel_size,
                 ),
             )
+        search, search_evidence = resolve_auto_high_search(
+            search=search,
+            sampling_us=_sampling_us_for_study_source(
+                study=study,
+                study_spec_source_path=self.study_root(study.study_id) / "study_spec.source",
+            ),
+        )
         spec = TrialSpec(
             study_id=study.study_id,
             trial_id=trial_id,
@@ -105,6 +162,7 @@ class StudyStore:
             probe_log_path=str(trial_root / "probe_history.json"),
             engine_log_path=str(trial_root / "engine.log"),
             result_path=str(trial_root / "result.json"),
+            search_evidence=search_evidence,
         )
         self.write_json(trial_root / "trial_spec.json", to_jsonable(spec))
         next_trial = (
@@ -251,47 +309,6 @@ def _parallel_size_for_proposal(*, study: StudySpec, proposal: Proposal) -> int:
     return _parallel_size_for_config(study=study, flag_patch=proposal.config_patch.flag_patch)
 
 
-def _inherit_incumbent_topology_for_runtime_patch(
-    *,
-    study: StudySpec,
-    state: StudyState,
-    proposal: Proposal,
-) -> Proposal:
-    flag_patch = dict(proposal.config_patch.flag_patch)
-    env_patch = dict(proposal.config_patch.env_patch)
-    if not flag_patch and not env_patch:
-        return proposal
-    if _TOPOLOGY_FLAG_KEYS.intersection(flag_patch):
-        return proposal
-    if not state.best_trial_id:
-        return proposal
-    incumbent = next(
-        (trial for trial in state.trials if trial.trial_id == state.best_trial_id),
-        None,
-    )
-    if incumbent is None or not isinstance(incumbent.config_patch, dict):
-        return proposal
-    incumbent_patch = incumbent.config_patch.get("flag_patch")
-    if not isinstance(incumbent_patch, dict):
-        return proposal
-    inherited_topology = {
-        key: value
-        for key, value in incumbent_patch.items()
-        if key in _TOPOLOGY_FLAG_KEYS and study.engine.base_flags.get(key) != value
-    }
-    if not inherited_topology:
-        return proposal
-    merged_flag_patch = dict(inherited_topology)
-    merged_flag_patch.update(flag_patch)
-    return replace(
-        proposal,
-        config_patch=ConfigPatch(
-            env_patch=env_patch,
-            flag_patch=merged_flag_patch,
-        ),
-    )
-
-
 def _parallel_size_for_trial_id(*, study: StudySpec, study_root: Path, trial_id: str) -> int | None:
     trial_spec_path = study_root / "trials" / trial_id / "trial_spec.json"
     if not trial_spec_path.exists():
@@ -323,3 +340,58 @@ def _derive_search_floor(*, study: StudySpec, state: StudyState, parallel_size:
     else:
         candidate = low
     return min(high, max(low, candidate))
+
+
+def _sampling_us_for_study_source(
+    *,
+    study: StudySpec,
+    study_spec_source_path: Path,
+) -> list[float]:
+    if not study.search.auto_high.enabled:
+        return []
+    from .trace import load_trace_requests
+
+    study_spec_path = Path(study_spec_source_path.read_text(encoding="utf-8").strip())
+    _, requests = load_trace_requests(study, study_spec_path=study_spec_path)
+    return [float(request.sampling_u) for request in requests]
+
+
+def resolve_auto_high_search(
+    *,
+    search: SamplingSearchSpec,
+    sampling_us: list[float],
+) -> tuple[SamplingSearchSpec, dict[str, Any]]:
+    policy = search.auto_high
+    trace_max_sampling_u = max(sampling_us) if sampling_us else None
+    evidence = {
+        "enabled": policy.enabled,
+        "original_high": search.high,
+        "effective_high": search.high,
+        "trace_max_sampling_u": trace_max_sampling_u,
+        "max_sampling_u": policy.max_sampling_u,
+        "require_human_confirmation_beyond_trace": (
+            policy.require_human_confirmation_beyond_trace
+        ),
+        "reason": "auto_high_disabled",
+    }
+    if not policy.enabled:
+        return search, evidence
+    if trace_max_sampling_u is None:
+        evidence["reason"] = "trace_has_no_sampling_u"
+        return search, evidence
+    ceiling = min(float(policy.max_sampling_u), 1.0, float(trace_max_sampling_u))
+    evidence["effective_ceiling"] = ceiling
+    if ceiling < float(search.low):
+        evidence["reason"] = "auto_high_ceiling_below_search_low"
+        return search, evidence
+    if abs(float(search.high) - ceiling) <= 1e-12:
+        evidence["reason"] = "search_high_already_at_auto_high_ceiling"
+        return search, evidence
+    updated = replace(search, high=ceiling)
+    evidence["effective_high"] = updated.high
+    evidence["reason"] = (
+        "search_high_raised_to_trace_ceiling"
+        if float(search.high) < ceiling
+        else "search_high_lowered_to_trace_ceiling"
+    )
+    return updated, evidence

src/aituner/worker.py

@@ -96,6 +96,7 @@ def _trial_spec_from_json(path: Path) -> TrialSpec:
         probe_log_path=str(payload["probe_log_path"]),
         engine_log_path=str(payload["engine_log_path"]),
         result_path=str(payload["result_path"]),
+        search_evidence=dict(payload.get("search_evidence") or {}),
     )
 
 
@@ -355,6 +356,59 @@ def _best_feasible_probe_record(probe_history: list[dict[str, Any]]) -> dict[str
     return max(feasible, key=lambda item: float(item["request_rate"]))
 
 
+def _binary_probe_resolution(search: SamplingSearchSpec) -> float:
+    return max(
+        float(search.tolerance),
+        (float(search.high) - float(search.low)) / float(2 ** max(search.max_probes, 1)),
+    )
+
+
+def _measurement_ceiling_evidence(
+    *,
+    search: SamplingSearchSpec,
+    requests: list[TraceRequest],
+    best_threshold: float | None,
+    best_payload: ProbePayload | None,
+) -> dict[str, Any]:
+    trace_max_sampling_u = max((float(request.sampling_u) for request in requests), default=None)
+    policy = search.auto_high
+    evidence: dict[str, Any] = {
+        "auto_high": {
+            "enabled": policy.enabled,
+            "max_sampling_u": policy.max_sampling_u,
+            "require_human_confirmation_beyond_trace": (
+                policy.require_human_confirmation_beyond_trace
+            ),
+        },
+        "search_high": search.high,
+        "trace_max_sampling_u": trace_max_sampling_u,
+        "measurement_ceiling_insufficient": False,
+        "reason": "measurement_ceiling_not_reached",
+    }
+    if trace_max_sampling_u is None:
+        evidence["reason"] = "trace_has_no_requests"
+        return evidence
+    if best_threshold is None or best_payload is None:
+        evidence["reason"] = "no_feasible_probe"
+        return evidence
+    resolution = _binary_probe_resolution(search)
+    threshold_gap_to_high = float(search.high) - float(best_threshold)
+    evidence["best_sampling_u"] = best_threshold
+    evidence["best_request_count"] = best_payload.request_count
+    evidence["threshold_gap_to_high"] = threshold_gap_to_high
+    evidence["binary_probe_resolution"] = resolution
+    full_trace_selected = best_payload.request_count >= len(requests)
+    high_reaches_trace = float(search.high) + 1e-12 >= float(trace_max_sampling_u)
+    if (
+        full_trace_selected
+        and high_reaches_trace
+        and threshold_gap_to_high <= resolution + 1e-12
+    ):
+        evidence["measurement_ceiling_insufficient"] = True
+        evidence["reason"] = "measurement_ceiling_insufficient"
+    return evidence
+
+
 def _replay_requests(
     requests: list[TraceRequest],
     *,
@@ -822,11 +876,19 @@ def run_trial(trial_spec_path: Path) -> dict[str, Any]:
                 *primary_search.probes,
                 *((fallback_search.probes if fallback_search is not None else [])),
             ]
+            measurement = _measurement_ceiling_evidence(
+                search=trial.search,
+                requests=requests,
+                best_threshold=search_for_best.best_threshold if best is not None else None,
+                best_payload=best,
+            )
+            measurement["auto_high_resolution"] = trial.search_evidence
             result = {
                 "study_id": trial.study_id,
                 "trial_id": trial.trial_id,
                 "status": "completed",
                 "config_patch": to_jsonable(trial.config_patch),
+                "measurement": measurement,
                 "best_source": best_source,
                 "best_sampling_u": search_for_best.best_threshold if best is not None else None,
                 "best_request_rate": best.request_rate if best is not None else None,

tests/test_declarative_harness.py

@@ -0,0 +1,156 @@
+from __future__ import annotations
+
+import unittest
+
+from aituner.declarative_harness import (
+    AxisSpec,
+    CoverageState,
+    HarnessPolicy,
+    OperatorSpec,
+    config_signature,
+    coverage_unit_id,
+    enumerate_candidate_set,
+    ordered_lattice_failure_region,
+    validate_coverage_stop,
+)
+
+
+class DeclarativeHarnessTests(unittest.TestCase):
+    def test_same_state_grammar_policy_candidate_set_is_deterministic(self) -> None:
+        axes = (
+            AxisSpec(name="tp", kind="ordered_lattice", values=(1, 2, 4)),
+            AxisSpec(name="gmu", kind="bounded_numeric", floor=0.7, ceiling=0.95, step=0.05),
+        )
+        policy = HarnessPolicy(
+            operators=(
+                OperatorSpec(name="runtime_climb", axis="gmu", kind="local_climb", harness_priority=1),
+                OperatorSpec(name="topology_bracket", axis="tp", kind="bracket", harness_priority=5),
+                OperatorSpec(name="runtime_floor", axis="gmu", kind="jump_to_floor", harness_priority=2),
+            )
+        )
+
+        first = enumerate_candidate_set({"tp": 2, "gmu": 0.8}, axes, policy)
+        second = enumerate_candidate_set({"gmu": 0.8, "tp": 2}, axes, policy)
+
+        self.assertEqual(first.candidate_set_hash, second.candidate_set_hash)
+        self.assertEqual(
+            [candidate.action_id for candidate in first.eligible],
+            [candidate.action_id for candidate in second.eligible],
+        )
+        self.assertEqual(
+            [blocked.reason for blocked in first.blocked],
+            [blocked.reason for blocked in second.blocked],
+        )
+        self.assertTrue(all(candidate.planner_score is None for candidate in first.eligible))
+        self.assertTrue(all(candidate.backend_score is None for candidate in first.eligible))
+
+    def test_toy_lattice_bracket_enumerates_all_other_lattice_points(self) -> None:
+        axis = AxisSpec(name="tp", kind="ordered_lattice", values=(1, 2, 4, 8))
+        policy = HarnessPolicy(
+            operators=(OperatorSpec(name="topology_bracket", axis="tp", kind="bracket"),)
+        )
+
+        candidate_set = enumerate_candidate_set({"tp": 2}, (axis,), policy)
+
+        self.assertEqual({candidate.target_value for candidate in candidate_set.eligible}, {1, 4, 8})
+        self.assertEqual(candidate_set.blocked, ())
+
+    def test_no_repeat_blocks_exact_candidate_signature_and_records_reason(self) -> None:
+        axis = AxisSpec(name="tp", kind="ordered_lattice", values=(1, 2, 4))
+        policy = HarnessPolicy(operators=(OperatorSpec(name="step", axis="tp", kind="step_up"),))
+        tested = CoverageState(tested_signatures=frozenset({config_signature({"tp": 4})}))
+
+        candidate_set = enumerate_candidate_set({"tp": 2}, (axis,), policy, tested)
+
+        self.assertEqual(candidate_set.eligible, ())
+        self.assertEqual(len(candidate_set.blocked), 1)
+        self.assertEqual(candidate_set.blocked[0].candidate.target_value, 4)
+        self.assertEqual(candidate_set.blocked[0].reason, "no_repeat: signature already tested")
+
+    def test_ordered_lattice_upper_boundary_uses_axis_values_not_hard_coded_tp8(self) -> None:
+        for values in ((1, 3, 9), (2, 5, 10, 20)):
+            with self.subTest(values=values):
+                axis = AxisSpec(name="parallel_size", kind="ordered_lattice", values=values)
+                policy = HarnessPolicy(
+                    operators=(OperatorSpec(name="step", axis=axis.name, kind="step_up"),)
+                )
+
+                candidate_set = enumerate_candidate_set({axis.name: values[-1]}, (axis,), policy)
+
+                self.assertEqual(candidate_set.eligible, ())
+                self.assertEqual(len(candidate_set.blocked), 1)
+                self.assertEqual(candidate_set.blocked[0].reason, "ordered_lattice_upper_boundary")
+                self.assertEqual(candidate_set.blocked[0].candidate.source_value, values[-1])
+
+    def test_bounded_numeric_jump_to_floor_uses_declared_floor_not_fixed_gmu_values(self) -> None:
+        for current, floor, ceiling in ((0.2, 0.6, 0.95), (0.77, 0.83, 0.91)):
+            with self.subTest(current=current, floor=floor, ceiling=ceiling):
+                axis = AxisSpec(
+                    name="memory_fraction",
+                    kind="bounded_numeric",
+                    floor=floor,
+                    ceiling=ceiling,
+                    step=0.02,
+                )
+                policy = HarnessPolicy(
+                    operators=(OperatorSpec(name="floor", axis="memory_fraction", kind="jump_to_floor"),)
+                )
+
+                candidate_set = enumerate_candidate_set({"memory_fraction": current}, (axis,), policy)
+
+                self.assertEqual(len(candidate_set.eligible), 1)
+                self.assertEqual(candidate_set.eligible[0].target_value, floor)
+                self.assertEqual(candidate_set.eligible[0].patch, {"memory_fraction": floor})
+
+    def test_coverage_stop_does_not_treat_signature_tested_as_coverage(self) -> None:
+        axis = AxisSpec(name="tp", kind="ordered_lattice", values=(1, 2))
+        required_unit = coverage_unit_id("tp", "step_up", 2)
+        policy = HarnessPolicy(
+            operators=(OperatorSpec(name="step", axis="tp", kind="step_up"),),
+            required_coverage_unit_ids=frozenset({required_unit}),
+        )
+        candidate = enumerate_candidate_set({"tp": 1}, (axis,), policy).eligible[0]
+        coverage_state = CoverageState(tested_signatures=frozenset({candidate.signature}))
+        candidate_set = enumerate_candidate_set({"tp": 1}, (axis,), policy, coverage_state)
+
+        stop = validate_coverage_stop(candidate_set, policy, coverage_state)
+
+        self.assertEqual(candidate_set.eligible, ())
+        self.assertEqual(stop.candidate_set_hash, candidate_set.candidate_set_hash)
+        self.assertFalse(stop.should_stop)
+        self.assertEqual(stop.reason, "coverage_units_missing")
+        self.assertEqual(stop.uncovered_unit_ids, (required_unit,))
+
+    def test_failure_invalidation_uses_conservative_region_not_exact_signature_only(self) -> None:
+        axis = AxisSpec(name="tp", kind="ordered_lattice", values=(1, 2, 4, 8))
+        policy = HarnessPolicy(
+            operators=(OperatorSpec(name="topology_bracket", axis="tp", kind="bracket"),)
+        )
+
+        exact_only = CoverageState(tested_signatures=frozenset({config_signature({"tp": 4})}))
+        exact_set = enumerate_candidate_set({"tp": 1}, (axis,), policy, exact_only)
+        self.assertEqual({candidate.target_value for candidate in exact_set.eligible}, {2, 8})
+
+        region = ordered_lattice_failure_region(
+            axis,
+            4,
+            direction="up",
+            reason="launch_failure_at_or_above_parallel_size",
+        )
+        regional_set = enumerate_candidate_set(
+            {"tp": 1},
+            (axis,),
+            policy,
+            CoverageState(failed_regions=(region,)),
+        )
+
+        self.assertEqual({candidate.target_value for candidate in regional_set.eligible}, {2})
+        blocked_targets = {blocked.candidate.target_value for blocked in regional_set.blocked}
+        self.assertTrue({4, 8}.issubset(blocked_targets))
+        self.assertTrue(
+            all("failure_region:tp:ge:4" in blocked.reason for blocked in regional_set.blocked)
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_interaction_matrix.py

@@ -0,0 +1,221 @@
+from __future__ import annotations
+
+import contextlib
+import io
+import json
+import tempfile
+import unittest
+from pathlib import Path
+
+from aituner.cli import main as cli_main
+from aituner.interaction_matrix import build_interaction_screening_matrix
+from aituner.spec import load_study_spec
+from aituner.trace import load_trace_requests, summarize_window
+
+
+def _write_interaction_study(tmp_path: Path) -> Path:
+    trace_dir = tmp_path / "trace_windows" / "traces"
+    trace_dir.mkdir(parents=True)
+    trace_path = trace_dir / "chat_w1.jsonl"
+    rows = []
+    for idx in range(24):
+        rows.append(
+            {
+                "request_id": f"r{idx}",
+                "timestamp": float(idx // 4),
+                "sampling_u": 0.1 + idx * 0.01,
+                "messages": [{"role": "user", "content": "hello"}],
+                "input_length": 7000 if idx < 20 else 9000,
+                "output_length": 64,
+            }
+        )
+    with trace_path.open("w", encoding="utf-8") as handle:
+        for row in rows:
+            handle.write(json.dumps(row) + "\n")
+
+    windows_path = tmp_path / "trace_windows" / "windows.json"
+    windows_path.write_text(
+        json.dumps(
+            {
+                "u_field": "sampling_u",
+                "windows": [
+                    {
+                        "window_id": "chat_w1",
+                        "trace_type": "chat",
+                        "trace_file": "traces/chat_w1.jsonl",
+                        "window_start": 0.0,
+                        "window_end": 6.0,
+                    }
+                ],
+            }
+        ),
+        encoding="utf-8",
+    )
+
+    study_path = tmp_path / "study.json"
+    study_path.write_text(
+        json.dumps(
+            {
+                "study_id": "interaction-study",
+                "hardware": {
+                    "gpu_count": 8,
+                    "gpu_model": "H20",
+                    "host_candidates": ["dash1"],
+                },
+                "model": {
+                    "model_id": "qwen",
+                    "served_model_name": "Qwen/Qwen3-30B-A3B-Instruct-2507",
+                },
+                "engine": {
+                    "engine_name": "vllm",
+                    "engine_version": "0.20",
+                    "exec_path": "/usr/local/bin/vllm",
+                    "cwd": str(tmp_path),
+                    "host": "127.0.0.1",
+                    "port": 8000,
+                    "healthcheck_path": "/v1/models",
+                    "ready_timeout_s": 30,
+                    "request_timeout_s": 30,
+                    "launch_args": ["serve", "/models/qwen"],
+                    "base_envs": {},
+                    "base_flags": {
+                        "host": "127.0.0.1",
+                        "port": 8000,
+                        "tensor-parallel-size": 2,
+                        "gpu-memory-utilization": 0.7,
+                    },
+                    "tunable_envs": [],
+                    "tunable_flags": [
+                        "tensor-parallel-size",
+                        "gpu-memory-utilization",
+                        "max-num-seqs",
+                        "max-num-batched-tokens",
+                    ],
+                    "topology_constraints": {
+                        "allowed_tensor_parallel_sizes": [1, 2, 4, 8],
+                        "allowed_tp_dp_products": [1, 2, 4, 8],
+                    },
+                    "python_executable": "python3",
+                },
+                "trace": {
+                    "windows_path": str(windows_path),
+                    "window_id": "chat_w1",
+                    "u_field": "sampling_u",
+                    "timestamp_field": "timestamp",
+                    "max_concurrency": 64,
+                },
+                "slo": {
+                    "target_pass_rate": 0.95,
+                    "ttft_rule": {"kind": "fixed_ms", "threshold_ms": 5000},
+                    "tpot_rule": {"kind": "fixed_ms", "threshold_ms": 120},
+                },
+                "search": {
+                    "low": 0.0,
+                    "high": 1.0,
+                    "tolerance": 0.01,
+                    "max_probes": 8,
+                    "sample_seed": 20260325,
+                },
+                "llm": {"system_prompt": "Tune it.", "max_history_trials": 8},
+            }
+        ),
+        encoding="utf-8",
+    )
+    return study_path
+
+
+class InteractionMatrixTests(unittest.TestCase):
+    def test_screening_matrix_uses_normalized_axis_scales(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            study_path = _write_interaction_study(Path(tmp))
+            study = load_study_spec(study_path)
+            window, requests = load_trace_requests(study, study_spec_path=study_path)
+
+            matrix = build_interaction_screening_matrix(
+                study=study,
+                window_summary=summarize_window(requests, window),
+            )
+
+            self.assertEqual(len(matrix["configs"]), 8)
+            self.assertEqual(
+                matrix["axes"]["tp"]["levels"],
+                {
+                    "low": {"tensor-parallel-size": 2},
+                    "high": {"tensor-parallel-size": 4},
+                },
+            )
+            self.assertEqual(matrix["axes"]["mns"]["levels"], {"low": 8, "high": 16})
+            self.assertEqual(matrix["axes"]["mbt"]["levels"], {"low": 9216, "high": 32768})
+            self.assertEqual(
+                matrix["fixed_tunable_base_flags"],
+                {"gpu-memory-utilization": 0.7},
+            )
+            patches = [
+                item["config_patch"]["flag_patch"]
+                for item in matrix["configs"]
+            ]
+            self.assertIn(
+                {
+                    "tensor-parallel-size": 4,
+                    "max-num-seqs": 16,
+                    "max-num-batched-tokens": 32768,
+                },
+                patches,
+            )
+
+    def test_screening_matrix_can_use_tp_dp_redistribution(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            study_path = _write_interaction_study(Path(tmp))
+            payload = json.loads(study_path.read_text(encoding="utf-8"))
+            payload["engine"]["base_flags"]["data-parallel-size"] = 2
+            payload["engine"]["tunable_flags"].append("data-parallel-size")
+            payload["engine"]["topology_constraints"] = {
+                "require_tp_dp_product_equals_gpu_count": True,
+                "allowed_tensor_parallel_sizes": [1, 2, 4, 8],
+                "allowed_data_parallel_sizes": [1, 2, 4, 8],
+            }
+            study_path.write_text(json.dumps(payload), encoding="utf-8")
+            study = load_study_spec(study_path)
+            window, requests = load_trace_requests(study, study_spec_path=study_path)
+
+            matrix = build_interaction_screening_matrix(
+                study=study,
+                window_summary=summarize_window(requests, window),
+            )
+
+            self.assertEqual(matrix["axes"]["tp"]["knob"], "topology")
+            self.assertEqual(
+                matrix["axes"]["tp"]["levels"],
+                {
+                    "low": {"tensor-parallel-size": 2, "data-parallel-size": 4},
+                    "high": {"tensor-parallel-size": 4, "data-parallel-size": 2},
+                },
+            )
+            self.assertNotIn("data-parallel-size", matrix["fixed_tunable_base_flags"])
+            self.assertIn(
+                {
+                    "tensor-parallel-size": 4,
+                    "data-parallel-size": 2,
+                    "max-num-seqs": 16,
+                    "max-num-batched-tokens": 32768,
+                },
+                [item["config_patch"]["flag_patch"] for item in matrix["configs"]],
+            )
+
+    def test_cli_profile_interaction_matrix_prints_json(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            study_path = _write_interaction_study(Path(tmp))
+            stdout = io.StringIO()
+
+            with contextlib.redirect_stdout(stdout):
+                rc = cli_main(["profile", "interaction-matrix", "--spec", str(study_path)])
+
+            self.assertEqual(rc, 0)
+            payload = json.loads(stdout.getvalue())
+            self.assertEqual(payload["matrix_kind"], "low_high_factorial_screening")
+            self.assertEqual(payload["source"]["window_id"], "chat_w1")
+            self.assertEqual(len(payload["configs"]), 8)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_mechanism_planner.py

@@ -0,0 +1,85 @@
+from __future__ import annotations
+
+import unittest
+
+from aituner.engine_adapters.vllm import default_vllm_descriptors
+from aituner.knob_descriptor import KnobConstraints, KnobDescriptor
+from aituner.mechanism_planner import (
+    CoordinateSearchPolicy,
+    coordinate_line_search_candidates,
+)
+
+
+class MechanismPlannerTests(unittest.TestCase):
+    def test_coordinate_search_uses_mechanism_not_knob_name(self) -> None:
+        vllm_descriptor = default_vllm_descriptors(tunable_flags=("max-num-seqs",))[0]
+        sglang_descriptor = KnobDescriptor(
+            name="max-running-requests",
+            location="flag",
+            value_type="int",
+            mechanisms=("admission_capacity", "kv_memory_pressure"),
+            search_geometry="positive_capacity",
+            operators=("coordinate_line_search",),
+            constraints=KnobConstraints(min_value=1, integer=True, multiple_of=8),
+            directional_effects={
+                "increase": ("admission_capacity",),
+                "decrease": ("kv_memory_pressure",),
+            },
+        )
+
+        vllm_candidates = coordinate_line_search_candidates(
+            current_config={"max-num-seqs": 8},
+            descriptors=(vllm_descriptor,),
+            evidence_weights={"admission_capacity": 0.9},
+        )
+        sglang_candidates = coordinate_line_search_candidates(
+            current_config={"max-running-requests": 8},
+            descriptors=(sglang_descriptor,),
+            evidence_weights={"admission_capacity": 0.9},
+        )
+
+        self.assertEqual(vllm_candidates[0].patch, {"max-num-seqs": 16})
+        self.assertEqual(sglang_candidates[0].patch, {"max-running-requests": 16})
+        self.assertEqual(vllm_candidates[0].mechanism, "admission_capacity")
+        self.assertEqual(sglang_candidates[0].mechanism, "admission_capacity")
+
+    def test_positive_capacity_can_decrease_for_memory_pressure(self) -> None:
+        descriptor = default_vllm_descriptors(tunable_flags=("max-num-seqs",))[0]
+
+        candidates = coordinate_line_search_candidates(
+            current_config={"max-num-seqs": 64},
+            descriptors=(descriptor,),
+            evidence_weights={"kv_memory_pressure": 0.8},
+        )
+
+        self.assertEqual(candidates[0].direction, "decrease")
+        self.assertEqual(candidates[0].patch, {"max-num-seqs": 32})
+
+    def test_bounded_fraction_respects_constraints(self) -> None:
+        descriptor = default_vllm_descriptors(tunable_flags=("gpu-memory-utilization",))[0]
+
+        candidates = coordinate_line_search_candidates(
+            current_config={"gpu-memory-utilization": 0.96},
+            descriptors=(descriptor,),
+            evidence_weights={"kv_memory_capacity": 0.8},
+        )
+
+        self.assertEqual(candidates[0].patch, {"gpu-memory-utilization": 0.97})
+
+    def test_coordinate_search_can_emit_larger_same_operator_steps(self) -> None:
+        descriptor = default_vllm_descriptors(tunable_flags=("max-num-seqs",))[0]
+
+        candidates = coordinate_line_search_candidates(
+            current_config={"max-num-seqs": 8},
+            descriptors=(descriptor,),
+            evidence_weights={"admission_capacity": 0.9},
+            policy=CoordinateSearchPolicy(step_multipliers=(1.0, 2.0)),
+        )
+
+        patches = [candidate.patch for candidate in candidates]
+        self.assertIn({"max-num-seqs": 16}, patches)
+        self.assertIn({"max-num-seqs": 24}, patches)
+
+
+if __name__ == "__main__":
+    unittest.main()