fix: record concurrency as a run condition, not a plan field

README.md

@@ -115,6 +115,14 @@ reproducible) and `contract.json` with the SHA-256 of those three files **and of
 re-checks all of it, so the scorer cannot change silently between freezing a plan and scoring a run, and a
 modified prompt or file is refused.
 
+`contract.generation` holds only what a request must carry and how output must be treated (temperature, seed,
+max_tokens, retries, truncation). Run conditions are deliberately absent: a plan cannot enforce how many requests a
+client keeps in flight, so `contract_sha256` stays a statement about the evaluation and the run records the
+condition it actually ran under. Concurrency lives in `run.<ts>.<seq>.json` next to `expected_concurrency`; pass
+`--expected-concurrency N` to have a differing `--concurrency` written into `deviations` as a structured
+`{field, expected, actual, source}` entry. A resumed run that changes concurrency is always recorded, whether or not
+a target was given.
+
 Plans frozen by the earlier upstream adapter (`contract_version: pyramidkv-mmlu-pro-subset-v1`) load only with
 `--allow-legacy-plan`; the scorer-integrity check cannot be applied to them and the reports say so
 (`legacy_plan: true`).

docs/QUICKSTART.md

@@ -51,6 +51,9 @@ hust-eval-mmlu-pro mmlu-pro prepare --assets <parquet目录> --model-path <模
 ```
 
 - 最后应打印 `12032 tasks of 12032`。想先试个子集,加 `--per-category 5`。
+- plan 里**没有**并发这一项:冻结的工件无法强制"客户端同时发几条",所以并发属于这一次运行的记录
+  (`run.<ts>.<seq>.json` 里的 `concurrency`),而不属于评测定义。想让工具替你核对,`run` 加
+  `--expected-concurrency N`;`--concurrency` 与之不同时会写进 `deviations`。
 - 输出目录必须不存在,已存在会报错。
 - **核对 `plan/cases.json` 的 SHA-256**(`hust-eval-mmlu-pro hash-file plan/cases.json`)。同样的模型、数据和参考,
   我们第一次冻结的结果是 `f77a7e5e6fd96fbbf46be7ab11f360309554105e97921805f02d1c93fc05fcf3`。

src/hust_eval_mmlu_pro/cli.py

@@ -6,6 +6,7 @@
 import sys
 
 from . import __version__
+from .mmlu_pro.run import DEFAULT_CONCURRENCY
 
 
 def _print(obj) -> None:
@@ -65,7 +66,7 @@ def _cmd_prepare(a):
 def _cmd_run(a):
     from .mmlu_pro.run import run
     s = run(a.plan, a.output, a.base_url, a.served_model, a.concurrency, a.timeout, a.limit, a.resume,
-            a.api_key_env, a.allow_legacy_plan, log=lambda m: print(m, flush=True))
+            a.api_key_env, a.allow_legacy_plan, a.expected_concurrency, log=lambda m: print(m, flush=True))
     _print(s)
     if s["transport_failures"]:
         sys.exit("transport failures present (kept, scored incorrect) - inspect requests/*.json")
@@ -168,7 +169,11 @@ def build_parser() -> argparse.ArgumentParser:
     s.add_argument("--base-url", default="http://127.0.0.1:8000")
     s.add_argument("--served-model", required=True)
     s.add_argument("--output", required=True)
-    s.add_argument("--concurrency", type=int, default=16)
+    s.add_argument("--concurrency", type=int, default=DEFAULT_CONCURRENCY, help="requests kept in flight (default: %d)"
+                   % DEFAULT_CONCURRENCY)
+    s.add_argument("--expected-concurrency", type=int,
+                   help="optional: the concurrency this run was planned for; a different --concurrency is then "
+                        "recorded in run.<ts>.<seq>.json -> deviations (default: not checked)")
     s.add_argument("--timeout", type=int, default=1800)
     s.add_argument("--limit", type=int, help="smoke test: first N cases only (not a score)")
     s.add_argument("--resume", action="store_true")

src/hust_eval_mmlu_pro/mmlu_pro/plan.py

@@ -129,8 +129,10 @@ def write_plan(output, inventory: List[dict], cases: List[dict], extra: Optional
                   "no random guessing",
         "scorer_sha256": sha256_file(scoring.__file__),
         "prompt_builder_sha256": sha256_file(prompt_mod.__file__),
+        # what a request must contain and how it must be treated; concurrency is NOT here, it is a property of a
+        # run (it cannot be enforced by the plan), so it is recorded in run.<ts>.<seq>.json instead
         "generation": {"temperature": 0, "seed": 17, "thinking": False, "max_tokens": max_output, "stop": None,
-                       "retries": 0, "concurrency": 1, "tools": [], "truncation": "reject"},
+                       "retries": 0, "tools": [], "truncation": "reject"},
         "prompt_contract": "reference five-shot CoT text wrapped in the model chat template, thinking disabled; "
                            "all five shots kept; no 'Question:' stop string; over-long prompts rejected",
         "files_sha256": {name: sha256_file(out / name) for name in PLAN_FILES},
@@ -171,7 +173,11 @@ def load_instruction(reference) -> str:
 
 def prepare(assets, model_path, reference, output, runtime_contract: Optional[str] = None,
             per_category: Optional[int] = None) -> dict:
-    """Freeze a plan from the pinned dataset files. Full test set unless per_category is given."""
+    """Freeze a plan from the pinned dataset files. Full test set unless per_category is given.
+
+    The plan describes what is evaluated (prompts, gold answers, limits, scorer) and the request parameters, but
+    not run conditions such as concurrency, which a plan cannot enforce; those live in the run's own record.
+    """
     from transformers import AutoTokenizer  # optional dependency
 
     model_path = Path(model_path)

src/hust_eval_mmlu_pro/mmlu_pro/run.py

@@ -3,7 +3,10 @@
 This is a pure client: streaming POST /v1/completions with the frozen token IDs of a plan, no retries (a failed
 request stays in the denominator and is scored incorrect), scoring with `scoring.score_results`.
 
-Deviation recorded in run.<ts>.json: concurrency is configurable, while the frozen plan declares 1.
+Deviations recorded in run.<ts>.<seq>.json are structured ({field, expected, actual, source}), never free text:
+the actual concurrency when it differs from an explicitly requested target, and concurrency drift between resumed
+invocations of the same run. The plan does not declare concurrency - a frozen artifact cannot enforce how many
+requests a client keeps in flight - so the run record is where a run condition belongs.
 """
 from __future__ import annotations
 
@@ -23,6 +26,8 @@
 
 _OPENER = urllib.request.build_opener(urllib.request.ProxyHandler({}))
 
+DEFAULT_CONCURRENCY = 16   # client-side default only; the plan deliberately says nothing about concurrency
+
 
 def one_request(base_url: str, model: str, case: dict, gen: dict, timeout: int, api_key: Optional[str] = None) -> dict:
     body = {
@@ -73,11 +78,14 @@ def one_request(base_url: str, model: str, case: dict, gen: dict, timeout: int,
 
 
 def run(plan: str, output: str, base_url: str = "http://127.0.0.1:8000",
-        served_model: str = "model", concurrency: int = 16, timeout: int = 1800, limit: Optional[int] = None,
-        resume: bool = False, api_key_env: Optional[str] = None, allow_legacy: bool = False, log=print) -> dict:
+        served_model: str = "model", concurrency: int = DEFAULT_CONCURRENCY, timeout: int = 1800,
+        limit: Optional[int] = None, resume: bool = False, api_key_env: Optional[str] = None,
+        allow_legacy: bool = False, expected_concurrency: Optional[int] = None, log=print) -> dict:
     api_key = os.environ.get(api_key_env) if api_key_env else None
     if api_key_env and not api_key:
         raise RuntimeError("environment variable %s is not set" % api_key_env)
+    if concurrency < 1:
+        raise ValueError("concurrency must be >= 1")
 
     contract, cases = plan_mod.load_plan(Path(plan), allow_legacy=allow_legacy)  # scorer, file and prompt hashes
     gen = contract["generation"]
@@ -88,20 +96,33 @@ def run(plan: str, output: str, base_url: str = "http://127.0.0.1:8000",
         raise FileExistsError("output exists; use a new directory (or resume=True to continue the same run)")
     (out / "requests").mkdir(parents=True, exist_ok=True)
 
+    deviations = []
+    if expected_concurrency is not None and concurrency != expected_concurrency:
+        deviations.append({"field": "concurrency", "expected": expected_concurrency, "actual": concurrency,
+                           "source": "--expected-concurrency"})
+    resumed_from = sorted(out.glob("run.*.json"))
+    for prev in resumed_from:   # one run directory must not silently mix two run conditions
+        p = json.loads(prev.read_text())
+        if p.get("concurrency") is not None and p["concurrency"] != concurrency:
+            deviations.append({"field": "concurrency", "expected": p["concurrency"], "actual": concurrency,
+                               "source": "earlier invocation " + prev.name})
+
     meta = {
         "tool": "hust-eval-mmlu-pro %s" % __version__,
         "runner_sha256": sha256_file(__file__),
         "scorer_sha256": sha256_file(scoring.__file__),
         "legacy_plan": bool(contract.get("_legacy_plan")),
         "contract_sha256": sha256_file(Path(plan) / "contract.json"),
-        "concurrency": concurrency, "upstream_concurrency": gen.get("concurrency"), "generation": gen,
+        "concurrency": concurrency, "expected_concurrency": expected_concurrency, "generation": gen,
         "base_url": base_url, "served_model": served_model, "n_cases": len(cases), "limit": limit,
         "resume": resume, "started_unix": time.time(),
-        "deviations": ["concurrency differs from contract.generation.concurrency"],
+        "deviations": deviations,
     }
     if limit:
         meta["WARNING"] = "smoke test only - not a score"
-    (out / ("run.%d.json" % int(time.time()))).write_text(json.dumps(meta, indent=1))
+    # the sequence keeps two invocations within the same second apart; names still sort chronologically
+    meta_path = out / ("run.%d.%03d.json" % (int(time.time()), len(resumed_from) + 1))
+    meta_path.write_text(json.dumps(meta, indent=1))
 
     done, lock, todo = {}, threading.Lock(), []
     for c in cases:

src/hust_eval_mmlu_pro/mmlu_pro/summary.py

@@ -78,11 +78,22 @@ def acc(group):
     }
     runs = sorted(glob.glob(os.path.join(accuracy_dir, "run.*.json")))
     if runs:
-        with open(runs[0]) as f:
-            meta = json.load(f)
-        report["run_meta"] = {k: meta.get(k) for k in ("concurrency", "upstream_concurrency", "generation", "n_cases",
+        metas = []
+        for fn in runs:
+            with open(fn) as f:
+                metas.append(json.load(f))
+        meta = metas[0]   # the invocation that started the run; any later one is a --resume
+        report["run_meta"] = {k: meta.get(k) for k in ("concurrency", "expected_concurrency", "generation", "n_cases",
                                                       "runner_sha256", "scorer_sha256", "adapter_sha256",
-                                                      "contract_sha256", "deviations")}
+                                                      "contract_sha256")}
+        report["run_meta"]["invocations"] = len(metas)
+        # deviations live in the invocation that saw them, so a resumed run must not hide them
+        deviations = []
+        for m in metas:
+            for d in m.get("deviations") or []:
+                if d not in deviations:
+                    deviations.append(d)
+        report["run_meta"]["deviations"] = deviations
     if bench_json and os.path.exists(bench_json):
         with open(bench_json) as f:
             b = json.load(f)

tests/test_client_flow.py

@@ -13,7 +13,7 @@
 
 
 class RunTests(unittest.TestCase):
-    def test_run_scores_and_records_deviations(self):
+    def test_run_scores_without_inventing_a_deviation(self):
         with tempfile.TemporaryDirectory() as d, StubServer() as srv:
             plan = make_plan(Path(d))
             out = Path(d) / "out"
@@ -26,13 +26,61 @@ def test_run_scores_and_records_deviations(self):
             self.assertTrue(srv.last_body["stream"])
             meta = json.loads(next(out.glob("run.*.json")).read_text())
             self.assertEqual(meta["concurrency"], 3)
-            self.assertEqual(meta["upstream_concurrency"], 1)
-            self.assertTrue(meta["deviations"])
+            self.assertIsNone(meta["expected_concurrency"])  # no target requested, so nothing to deviate from
+            self.assertEqual(meta["deviations"], [])
+            self.assertNotIn("concurrency", meta["generation"])  # a plan cannot declare a run condition
             self.assertEqual(len(meta["scorer_sha256"]), 64)
             self.assertFalse(meta["legacy_plan"])
             with self.assertRaises(FileExistsError):
                 run.run(str(plan), str(out), srv.url, "stub-model", log=lambda m: None)
 
+    def test_expected_concurrency_mismatch_is_structured(self):
+        with tempfile.TemporaryDirectory() as d, StubServer() as srv:
+            plan = make_plan(Path(d))
+            out = Path(d) / "out"
+            run.run(str(plan), str(out), srv.url, "stub-model", concurrency=3, expected_concurrency=1,
+                    log=lambda m: None)
+            meta = json.loads(next(out.glob("run.*.json")).read_text())
+            self.assertEqual(meta["deviations"], [{"field": "concurrency", "expected": 1, "actual": 3,
+                                                  "source": "--expected-concurrency"}])
+            rep = summary.build_report(str(plan), str(out))
+            self.assertEqual(len(rep["run_meta"]["deviations"]), 1)
+
+    def test_matching_expected_concurrency_records_nothing(self):
+        with tempfile.TemporaryDirectory() as d, StubServer() as srv:
+            plan = make_plan(Path(d))
+            out = Path(d) / "out"
+            run.run(str(plan), str(out), srv.url, "stub-model", concurrency=3, expected_concurrency=3,
+                    log=lambda m: None)
+            meta = json.loads(next(out.glob("run.*.json")).read_text())
+            self.assertEqual(meta["expected_concurrency"], 3)
+            self.assertEqual(meta["deviations"], [])
+            rep = summary.build_report(str(plan), str(out))
+            self.assertEqual(rep["run_meta"]["deviations"], [])
+            self.assertEqual(rep["run_meta"]["invocations"], 1)
+
+    def test_resume_with_another_concurrency_is_reported(self):
+        with tempfile.TemporaryDirectory() as d, StubServer() as srv:
+            plan = make_plan(Path(d))
+            out = Path(d) / "out"
+            run.run(str(plan), str(out), srv.url, "stub-model", concurrency=4, limit=2, log=lambda m: None)
+            run.run(str(plan), str(out), srv.url, "stub-model", concurrency=1, resume=True, log=lambda m: None)
+            metas = [json.loads(p.read_text()) for p in sorted(out.glob("run.*.json"))]
+            self.assertEqual(len(metas), 2)  # two invocations inside one second still get their own file
+            self.assertEqual(metas[0]["deviations"], [])
+            self.assertEqual(metas[1]["deviations"], [{"field": "concurrency", "expected": 4, "actual": 1,
+                                                      "source": "earlier invocation " + sorted(
+                                                          out.glob("run.*.json"))[0].name}])
+            rep = summary.build_report(str(plan), str(out))
+            self.assertEqual(rep["run_meta"]["invocations"], 2)
+            self.assertEqual(len(rep["run_meta"]["deviations"]), 1)
+
+    def test_concurrency_below_one_is_rejected(self):
+        with tempfile.TemporaryDirectory() as d, StubServer() as srv:
+            plan = make_plan(Path(d))
+            with self.assertRaises(ValueError):
+                run.run(str(plan), str(Path(d) / "out"), srv.url, "stub-model", concurrency=0, log=lambda m: None)
+
     def test_failed_requests_are_kept_and_scored_incorrect(self):
         with tempfile.TemporaryDirectory() as d:
             plan = make_plan(Path(d))