harness2: upstream gap fill — trivial selection fix, session budget, run_tag validation, preflight check

- select_final trivial condition now k==1 and single-candidate pool,
  regardless of whether a selector is provided (previously gated on
  selector is None, so wired runs always paid for a useless selector
  task at k=1)
- Budget gains max_sessions gate (probe + stage-attempt sessions;
  fmt2 corrections and selector/reflector sessions exempt), the
  backstop for unpriced models where the USD gate never binds
- validate_run_tag (charset, no leading '-', no s<N> seed prefix)
  enforced at run_improvement_via_tasks and the CLI
- CLI --check preflight (base harness, source dir, prompt, tag,
  control-plane /health probe) without creating tasks
- ModeResult.to_dict now includes rollout summaries so a lost probe
  is visible in full_result instead of only on the server
- CLI plan counts the selector only when k>1

cloud/harness2.py

@@ -26,8 +26,9 @@ Claude CLI 通道都没有搬。
 
 选终版时只比“产物”：候选产物先匿名化（A/B/C……，按任务+k+模式
 定种子洗牌，可复现），选择器看不到哪个产物出自哪个 harness；选不
-出来时回落到已交付的终版，绝不抛异常把整轮炸掉。k=1 且没有选择
-器时直接走 trivial：唯一的终版就是答案，不必再选一次。
+出来时回落到已交付的终版，绝不抛异常把整轮炸掉。k=1 且终版池
+只有唯一候选时直接走 trivial：唯一的终版就是答案，不必再选
+一次（与选择器是否提供无关）。
 
 经验库（上游叫 harnessing-level recursion）：
   采纳后，反思器只写一条有证据的经验（cue / lesson / evidence 三项
@@ -740,16 +741,32 @@ def _mission_experience() -> str:
 
 @dataclass
 class Budget:
+    """美元门 + 会话数门（2026-10-29 对照补齐）。
+
+    上游用会话数记账并强校验期望计数（R=4k、P=3k）；本项目此前
+    只有美元门，在 unpriced 模型上成本恒为 0、门永远不触发。
+    sessions 计探针会话与阶段尝试会话；edit() 内部的 fmt2 纠正
+    与选择/反思会话不计入（前者在接线层、后者不走 budget），
+    上限语义因此是“受门控的会话数”，不是全轮任务总数。
+    """
+
     max_cost_usd: float | None = None
     spent_usd: float = 0.0
+    max_sessions: int | None = None
+    sessions: int = 0
 
     @property
     def exceeded(self) -> bool:
-        return self.max_cost_usd is not None and self.spent_usd > self.max_cost_usd
+        if self.max_cost_usd is not None and self.spent_usd > self.max_cost_usd:
+            return True
+        return self.max_sessions is not None and self.sessions >= self.max_sessions
 
     def add(self, cost: float) -> None:
         self.spent_usd += float(cost or 0.0)
 
+    def add_session(self) -> None:
+        self.sessions += 1
+
 
 # ---------------------------------------------------------------------------
 # 阶段：一次提议（含一次修正机会）+ 一次探针
@@ -816,6 +833,18 @@ class ModeResult:
             if any(s.dim == "AGG" for s in self.stages)
             else None,
             "errors": self.errors,
+            # 探针结果此前不在 full_result 里：real5 的终版探针没建
+            # 出来时，本地日志完全看不到原因，只能去服务端翻任务。
+            # 只放摘要（不放 trajectory），避免日志爆炸。
+            "rollouts": {
+                lbl: {
+                    "ok": r.ok,
+                    "delivered": r.delivered,
+                    "error": (r.error or "")[:400],
+                    "deliverable_text": (r.deliverable_text or "")[:500],
+                }
+                for lbl, r in self.rollouts.items()
+            },
         }
 
 
@@ -872,6 +901,11 @@ def _run_stage(
     err = ""
     spent = 0.0
     for attempt in range(1, STAGE_ATTEMPTS + 1):
+        if budget is not None:
+            if budget.exceeded:
+                err = "budget exceeded before this attempt"
+                break
+            budget.add_session()
         try:
             result = editor.edit(
                 full_mission + (
@@ -918,6 +952,8 @@ def _roll(
     if budget is not None and budget.exceeded:
         return Rollout(label=label, ok=False, delivered=False,
                        error="budget exceeded before this rollout")
+    if budget is not None:
+        budget.add_session()
     try:
         rv = adapter.execute(label, tree)
     except Exception as exc:
@@ -1164,8 +1200,11 @@ def select_final(
                   f"{text.count(chr(10)) + 1:,} lines. COMPLETE -- nothing trimmed. -->\n")
         (cand_dir / f"{anon}.md").write_text(header + text, encoding="utf-8")
     ids = ", ".join(sorted(sel.id_map))
-    if selector is None and k == 1:
-        # 唯一终版：不必再开一轮选择会话（上游的 trivial_selection）。
+    if k == 1 and len(pool) == 1:
+        # 唯一终版：不必再开一轮选择会话（上游的 trivial_selection，
+        # 条件是“终版池 + k=1”，与选择器是否提供无关；此前误写成
+        # selector is None 才 trivial，而接线永远传选择器，导致
+        # k=1 也白白跑一个选择任务——2026-10-29 对照发现）。
         label = next(iter(pool))
         sel.ok = True
         sel.trivial = True
@@ -1402,6 +1441,7 @@ def improve_task(
     library: Library | None = None,
     auto_apply: bool = False,
     budget_usd: float | None = None,
+    max_sessions: int | None = None,
 ) -> ImprovementResult:
     """跑一轮完整改进。base 探针先跑一次，作为全部候选共享的证据。
 
@@ -1413,7 +1453,7 @@ def improve_task(
     if not 1 <= int(k) <= MAX_K:
         raise ValueError(f"k must be 1..{MAX_K}, got {k!r}")
     work_dir = Path(work_dir)
-    budget = Budget(max_cost_usd=budget_usd)
+    budget = Budget(max_cost_usd=budget_usd, max_sessions=max_sessions)
     base_tree = adapter.base_harness()
     # 校验 base 本身：base 若已不合法，整轮的对比就没意义，直接报错
     # 而不是让候选在坏 base 上各跑各的。

cloud/harness2_tasks.py

@@ -62,6 +62,7 @@ parallel 一轮 = 1 base + 2 候选 + 1 终版探针 + 3 编辑 + 1 选择 +
 from __future__ import annotations
 
 import json
+import re
 import time
 import urllib.error
 import urllib.request
@@ -91,6 +92,67 @@ DEFAULT_TASK_TIMEOUT_SECONDS = 3600
 DEFAULT_POLL_SECONDS = 10
 TRACE_MAX_CHARS = 50000
 
+_TAG_RX = re.compile(r"[A-Za-z0-9_.][A-Za-z0-9_.\-]*\Z")
+_SEED_TAG_RX = re.compile(r"s\d+(-|\Z)")
+
+
+def validate_run_tag(tag: str) -> None:
+    """校验 run_tag 形状（上游 schedule.validate_tag 的对照补齐）。
+
+    run_tag 会进幂等键与路径段：空串表示“调用方稍后生成”，不
+    校验；非空时只允许字母数字与 _ . -、首字符不能是 -、不能
+    以 s + 数字开头（上游用 s<n> 段表示 seed，撞名会把两轮的
+    产物算到同一命名空间）。
+    """
+    if not tag:
+        return
+    if not _TAG_RX.match(tag):
+        raise ValueError(
+            f"run_tag {tag!r} must match [A-Za-z0-9_.-] with no leading '-' "
+            "and no '/'")
+    if _SEED_TAG_RX.match(tag):
+        raise ValueError(
+            f"run_tag {tag!r} must not start with a seed segment like 's1'")
+
+
+def preflight_checks(*, base_harness_dir: str | Path, source_dir: str | Path | None = None,
+                     prompt: str = "", run_tag: str = "",
+                     base_url: str | None = None) -> tuple[list[str], list[str]]:
+    """建任务前的本地 + 可达性预检（2026-10-29 对照补齐）。
+
+    返回 (errors, warnings)：errors 非空时调用方不得建任务；
+    warnings 只告警（health 探针失败、run_tag 为空待生成等不
+    拦任务）。本函数不读密钥、不建任务，--check 可离线跑。
+    """
+    errors: list[str] = []
+    warnings: list[str] = []
+    base_dir = Path(base_harness_dir)
+    agents = base_dir / "AGENTS.md"
+    if not base_dir.is_dir():
+        errors.append(f"base-harness-dir 不存在或不是目录：{base_harness_dir}")
+    elif not agents.is_file() or not agents.read_text(
+            encoding="utf-8", errors="replace").strip():
+        errors.append(f"base-harness-dir 里没有可用的 AGENTS.md：{base_harness_dir}")
+    if source_dir is not None and str(source_dir) != "":
+        if not Path(source_dir).is_dir():
+            errors.append(f"source-dir 不存在或不是目录：{source_dir}")
+    if not (prompt or "").strip():
+        errors.append("需要 --prompt 或 --prompt-file（任务提示为空）")
+    if run_tag:
+        try:
+            validate_run_tag(run_tag)
+        except ValueError as exc:
+            errors.append(str(exc))
+    if base_url:
+        try:
+            req = urllib.request.Request(base_url.rstrip("/") + "/health")
+            with urllib.request.urlopen(req, timeout=10) as resp:
+                if resp.status != 200:
+                    warnings.append(f"控制面 health 非 200：{resp.status}")
+        except Exception as exc:
+            warnings.append(f"控制面 health 探针失败（不拦任务）：{type(exc).__name__}: {exc}")
+    return errors, warnings
+
 
 def unwrap_result_text(raw: str) -> str:
     """把服务端 result_text 的 JSON 信封拆开，只取 agent 的最终回复。
@@ -709,6 +771,7 @@ def run_improvement_via_tasks(
     mode: str = "parallel",
     library_root: Path | None = None,
     budget_usd: float | None = None,
+    max_sessions: int | None = None,
     model: str | None = None,
     effort: str | None = None,
     approval: str | None = None,
@@ -739,10 +802,11 @@ def run_improvement_via_tasks(
     selector = CodexSelector(client=client, project_id=project_id, run_tag=run_tag, **shared)
     reflector = CodexReflector(client=client, project_id=project_id, run_tag=run_tag, **shared)
     library = Library(library_root, domain) if library_root is not None else None
+    validate_run_tag(run_tag)
     return improve_task(
         adapter, editor, selector=selector, reflector=reflector,
         work_dir=work_dir, k=k, mode=mode, library=library,
-        auto_apply=False, budget_usd=budget_usd)
+        auto_apply=False, budget_usd=budget_usd, max_sessions=max_sessions)
 
 
 __all__ = [
@@ -750,5 +814,5 @@ __all__ = [
     "TRACE_MAX_CHARS", "TaskState", "TaskClient", "build_multipart",
     "HttpTaskClient", "CompletedTask", "run_task",
     "CodexProbeAdapter", "CodexEditor", "CodexSelector", "CodexReflector",
-    "run_improvement_via_tasks",
+    "run_improvement_via_tasks", "validate_run_tag", "preflight_checks",
 ]

cloud/tests/test_harness2_tasks.py

@@ -24,8 +24,8 @@ try:
     )
     from cloud.harness2_tasks import (
         CodexEditor, CodexProbeAdapter, HttpTaskClient, TaskState,
-        build_multipart, run_improvement_via_tasks, run_task,
-        unwrap_result_text,
+        build_multipart, preflight_checks, run_improvement_via_tasks, run_task,
+        unwrap_result_text, validate_run_tag,
     )
     from cloud.uploads import parse_multipart, parse_paths_field
 except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
@@ -37,8 +37,8 @@ except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.p
     )
     from cloud.harness2_tasks import (
         CodexEditor, CodexProbeAdapter, HttpTaskClient, TaskState,
-        build_multipart, run_improvement_via_tasks, run_task,
-        unwrap_result_text,
+        build_multipart, preflight_checks, run_improvement_via_tasks, run_task,
+        unwrap_result_text, validate_run_tag,
     )
     from cloud.uploads import parse_multipart, parse_paths_field
 
@@ -211,7 +211,7 @@ class TestParseHelpers(unittest.TestCase):
 
 
 class TestFullRunViaTasks(unittest.TestCase):
-    def _run(self, client, run_tag="run1", work_sub="work"):
+    def _run(self, client, run_tag="run1", work_sub="work", k=1, max_sessions=None):
         tmp = tempfile.TemporaryDirectory()
         self.addCleanup(tmp.cleanup)
         root = Path(tmp.name)
@@ -221,7 +221,8 @@ class TestFullRunViaTasks(unittest.TestCase):
             base_harness=dict(BASE_TREE), run_tag=run_tag,
             work_dir=root / work_sub, domain="accounting",
             source_files={"ledger.csv": LEDGER},
-            k=1, mode="parallel", library_root=root / "library")
+            k=k, mode="parallel", library_root=root / "library",
+            max_sessions=max_sessions)
         return result, client, root
 
     def test_parallel_k1_via_real_task_shape(self):
@@ -229,10 +230,11 @@ class TestFullRunViaTasks(unittest.TestCase):
         result, _, root = self._run(client)
         self.assertTrue(result.ok)
         self.assertEqual(result.selection.chosen_label, "agg1")
+        self.assertTrue(result.selection.trivial)  # k=1 单终版不开选择会话
         self.assertEqual(result.mode_result.rollouts["agg1"].deliverable_text,
                          "Total: 85.15")
-        # 4 探针 + 3 编辑 + 1 选择 + 1 反思 = 9 个任务
-        self.assertEqual(client.created, 9)
+        # 4 探针 + 3 编辑 + 1 反思 = 8 个任务（选择 trivial，不建任务）
+        self.assertEqual(client.created, 8)
         self.assertIsNotNone(result.library_entry)
         self.assertTrue((root / "library" / "accounting" / "INDEX.md").is_file())
 
@@ -253,8 +255,9 @@ class TestFullRunViaTasks(unittest.TestCase):
         self.assertIsNone(result.library_entry)
 
     def test_selector_gibberish_falls_back(self):
+        # k=1 时选择 trivial，不走选择器；用 k=2 让选择器真正跑起来。
         client = FakeTaskClient(selector_mode="gibberish")
-        result, _, _ = self._run(client)
+        result, _, _ = self._run(client, k=2)
         self.assertTrue(result.ok)
         self.assertTrue(result.selection.fallback)
 
@@ -338,6 +341,79 @@ class TestEnvelopeFullRound(unittest.TestCase):
         self.assertIsNotNone(result.library_entry)
 
 
+class TestSessionBudget(unittest.TestCase):
+    """会话数门：0 挡全部、2 挡在中途；fmt2 与选择/反思不计入。"""
+
+    def _run(self, max_sessions):
+        client = FakeTaskClient()
+        tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(tmp.cleanup)
+        root = Path(tmp.name)
+        result = run_improvement_via_tasks(
+            client, project_id="p1", task_id="ledger-1",
+            prompt="Sum every signed amount in ledger.csv. Return Total: with two decimals.",
+            base_harness=dict(BASE_TREE), run_tag="sess",
+            work_dir=root / "work", domain="accounting",
+            source_files={"ledger.csv": LEDGER},
+            k=1, mode="parallel", library_root=root / "library",
+            max_sessions=max_sessions)
+        return result, client
+
+    def test_zero_blocks_everything(self):
+        result, client = self._run(0)
+        self.assertEqual(client.created, 0)
+        self.assertEqual(set(result.mode_result.fell_back), {"inv1", "dom1", "agg1"})
+
+    def test_mid_cap_stops_after_two_sessions(self):
+        # base 探针（1）+ inv1 编辑（2），之后 dom1/agg1 与探针全被门挡住。
+        result, client = self._run(2)
+        self.assertEqual(client.created, 2)
+        self.assertIn("dom1", result.mode_result.fell_back)
+        self.assertIn("agg1", result.mode_result.fell_back)
+        self.assertNotIn("inv1", result.mode_result.fell_back)
+
+
+class TestValidateRunTag(unittest.TestCase):
+    def test_valid(self):
+        for tag in ("h2-20261029-real1", "a", "a.b_c-1", "run_1"):
+            validate_run_tag(tag)
+
+    def test_invalid(self):
+        for tag in ("-bad", "has/slash", "has space", "s1-foo", "s12"):
+            with self.assertRaises(ValueError, msg=tag):
+                validate_run_tag(tag)
+        validate_run_tag("")  # 空串表示稍后生成，不校验
+
+
+class TestPreflight(unittest.TestCase):
+    def _dirs(self, agents="Use scripts/summarize.py.\n"):
+        tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(tmp.cleanup)
+        root = Path(tmp.name)
+        (root / "base").mkdir()
+        if agents is not None:
+            (root / "base" / "AGENTS.md").write_text(agents)
+        (root / "source").mkdir()
+        (root / "source" / "ledger.csv").write_bytes(LEDGER)
+        return root
+
+    def test_ok(self):
+        root = self._dirs()
+        errors, _ = preflight_checks(
+            base_harness_dir=root / "base", source_dir=root / "source",
+            prompt="do it", run_tag="h2-ok")
+        self.assertEqual(errors, [])
+
+    def test_errors(self):
+        root = self._dirs(agents="")
+        errors, _ = preflight_checks(
+            base_harness_dir=root / "base", source_dir=root / "missing",
+            prompt="  ", run_tag="s1-bad")
+        self.assertEqual(len(errors), 4)  # AGENTS 为空、source 缺失、提示为空、tag 非法
+        errors, _ = preflight_checks(base_harness_dir=root / "nope", prompt="x")
+        self.assertTrue(any("不存在" in e for e in errors))
+
+
 class TestHttpClient(unittest.TestCase):
     def test_client_against_local_server(self):
         seen = {"keys": [], "uploads": [], "tasks": []}