harness2: recursive harness optimization wired to real codex tasks

Port Google Research's Harness^2 method layer (model weights frozen,
only the surrounding harness is optimized): Propose-Probe-Compose with
INV/DOM/AGG editors, parallel/sequential modes, per-domain experience
library, anonymous-product selection, k<=3 hard cap and budget gate.
Execution goes only through the existing control-plane API (upload ->
task with idempotency key -> poll -> trace/cost -> cancel on timeout);
no new server endpoints, no codex flags composed, tasks inherit the
normal executor/sandbox/quota/audit path.

Wiring lessons from four real rounds against production (project
wwww, ledger task): HttpTaskClient retries transient disconnects and
502/503/429 (3 attempts); editors get a worked file-block example, an
explicit <no-change> marker and one feedback retry (.fmt2) instead of
silently accepting notes-only replies; run_task unwraps the server's
result_text JSON envelope (text field holds the real reply) — without
this, line-anchored file-block parsing could never see blocks the
agents were actually emitting. Final round (real4) fully green:
all stages landed, selection non-fallback, experience入库, probe
delivers the correct signed total 85.15.

Tests: 35 harness tests (17 method + 18 wiring incl. fake-control-plane
full round, idempotent replay, retry, format retry, envelope round).
Deliberate deviations from upstream are documented in docs/harness2.md
(no network-unrestricted/sandbox-disabled defaults, auto_apply off,
no benchmark/scorer, probe deliverable = final reply text).

cloud/harness2_tasks.py

@@ -0,0 +1,754 @@
+"""Harness² 接线层：把 cloud/harness2.py 的三个协议落到真实 codex 任务上。
+
+只用控制面已有的 API，不新增任何服务端端点：
+  POST /uploads（multipart + paths JSON）-> upload_id
+  POST /tasks（upload_id 绑定 + idempotency_key 幂等）-> 任务排队
+  GET  /tasks/{id}        轮询终态（completed / failed / interrupted /
+                          merged / merge-failed）
+  GET  /tasks/{id}/trace  轨迹（探针的 trajectory 来源）
+  GET  /tasks/{id}/cost   成本（usd_micros / 1e6；无价 token 不计入 USD，
+                          这是现有 cost 模型的边界，不是本层漏算）
+  POST /tasks/{id}/cancel 超时取消，不留孤儿任务继续烧钱
+认证：X-API-Key 请求头；密钥只由调用方持有，本模块不读、不存、
+不打印密钥。
+
+三类任务是怎么跑的：
+- 探针（CodexProbeAdapter.execute）：bundle = 源文件 + harness 树。
+  Worker 派发时把 bundle 经 host.py --seed-dir 拷进 worktree，harness
+  文件会覆盖仓库同名文件——这正是探针要的效果：同一任务在不同
+  harness 下跑。prompt = 原任务 prompt + 一段交付脚注（见
+  _PROBE_DELIVERABLE_FOOTER），因为评审只看最终回复：agent 必须
+  把交付物完整写在最终回复里，不能只说“文件已写好”。这就是本层
+  最大的适用边界：交付物是最终回复文本的任务；文件型产物（代码
+  diff、报告文件）目前只能靠最终回复复述/摘要，worktree 快照 API
+  只给 diff 截断与未跟踪文件列表、不给完整文件内容，跨主机还读
+  不到——文件产物评审需要新增文件读取端点，另行评审，不在本层
+  假装已解决。
+- 编辑（CodexEditor.edit）：bundle = context/ 前缀的证据树（task.md /
+  rollouts / deliverables / harness / library / pool / decisions）。
+  agent 在 worktree 的 context/harness/ 里改文件，最终回复里用
+  文件块只输出改动的文件（没改的省略即继承），外加 <notes> 块。
+  接线层解析文件块、落到该阶段的 fallback 树上，再走核心校验。
+  最终回复里既没有文件块也没有 <notes>，或文件块路径越界、未
+  闭合，都算该阶段失败：核心循环给一次修正机会，再失败回落到
+  fallback 树——绝不把“没看懂输出格式”静默当成“无需改动”。
+- 选择（CodexSelector.select）与反思（CodexReflector.reflect）：同款
+  bundle + 最终回复解析（```choice + <why>；<experience> + <edits>），
+  不读 worktree 文件。选择失败时核心循环回落到已交付终版；反思
+  三项不全时返回 {}，不入库。
+
+幂等与断点续跑：每个任务都带 idempotency_key =
+f"h2-{run_tag}-{role}-{label}"。同一 run_tag 重跑时服务端直接返回
+已有任务（200 而非 201），已终态的任务直接复用结果——这是断点
+续跑语义，不是 bug；想从头重跑必须换 run_tag。注意：换了 harness
+内容但沿用旧 run_tag 时，旧任务结果会被复用，改进轮次就白跑了，
+所以 run_tag 必须随“任务 + harness 基线 + 参数”一起变，调用方
+（见 examples/harness2_run.py）默认用时间戳生成，显式 --run-tag
+只用于“接着上次断点跑”。
+
+成本与审批：本层不拼任何 codex 参数、不改任务默认值。model /
+effort / approval / environment_id / executor 全部由调用方透传，
+缺省时走 Worker 现有默认值；探针与编辑任务走与普通任务完全相同
+的执行器、沙箱、租户隔离、配额门与审计。长任务走 #133 的自动续
+跑；接线层单任务超时默认 3600 秒，超时先取消任务再记失败（探针）
+或抛错走回落（编辑/选择/反思），不留任务在后台继续烧钱。
+
+本层是同步阻塞的：improve_task 每一步都要等任务终态。k=1
+parallel 一轮 = 1 base + 2 候选 + 1 终版探针 + 3 编辑 + 1 选择 +
+1 反思 ≈ 9 个任务串行等待，每个都可能几分钟——这是方法本身的
+成本（上游同款：4k 次执行 + 3k 次改进会话），不是本层低效。
+"""
+
+from __future__ import annotations
+
+import json
+import time
+import urllib.error
+import urllib.request
+import uuid
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Protocol
+
+from .harness2 import (
+    EditResult,
+    Rollout,
+    Tree,
+    improve_task,
+    parse_choice,
+    parse_experience,
+    parse_harness_files,
+    parse_no_change,
+    parse_notes,
+    parse_why,
+    read_tree,
+)
+
+TERMINAL_STATUSES = frozenset(
+    {"completed", "failed", "interrupted", "merged", "merge-failed"}
+)
+DEFAULT_TASK_TIMEOUT_SECONDS = 3600
+DEFAULT_POLL_SECONDS = 10
+TRACE_MAX_CHARS = 50000
+
+
+def unwrap_result_text(raw: str) -> str:
+    """把服务端 result_text 的 JSON 信封拆开，只取 agent 的最终回复。
+
+    2026-10-29 诊断的真根因：控制面返回的 result_text 不是 agent
+    原文，而是一层 JSON 信封（task / thread_id / cwd / status /
+    text / usage …），text 字段里才是最终回复。接线层此前原样把
+    信封当回复用：notes / choice / experience 的正则不锚定行首，
+    在信封里照样命中，所以三轮里只有行首锚定的文件块解析一直
+    看不见块——编辑明明发了块，却被判成 notes-only、无改动、
+    格式错误。拆信封放在 run_task 这一个 choke point：探针、
+    编辑、选择、反思四路共用它，一处拆全受益；拆不开（非
+    JSON、无 text 字段、text 非字符串）时原样返回，不丢内容。
+    """
+    if not raw or not raw.lstrip().startswith("{"):
+        return raw
+    try:
+        obj = json.loads(raw)
+    except ValueError:
+        return raw
+    if isinstance(obj, dict) and isinstance(obj.get("text"), str):
+        return obj["text"]
+    return raw
+
+_PROBE_DELIVERABLE_FOOTER = """
+
+[DELIVERABLE — READ THIS BEFORE YOU FINISH]
+Your final reply IS the deliverable that will be judged. When you finish,
+write the complete deliverable in your final reply itself — do not only say
+"done" or "the file is written". If the task asks for a file, also write the
+file in the worktree, but the final reply must still contain the full content
+a reviewer needs to judge the work.
+"""
+
+_EDITOR_OUTPUT_FORMAT = """
+
+# OUTPUT FORMAT — YOU MUST FOLLOW THIS EXACTLY
+Edit the files under context/harness/ in the worktree. Then, in your final
+reply, output ONLY the files you changed, one block per file, in this format
+(files you did not change are inherited automatically — do NOT re-emit them):
+
+=== HARNESS FILE: <relative path on the editable surface> ===
+<the file's COMPLETE new content>
+=== END HARNESS FILE ===
+
+Example (a one-line file changed in full — your blocks must contain the
+COMPLETE new content, not a diff and not a description of the change):
+
+=== HARNESS FILE: scripts/summarize.py ===
+print("Total: 85.15")
+=== END HARNESS FILE ===
+
+Then a <notes>...</notes> block: what you changed, and the exact trajectory
+moment that justifies each change.
+
+HARD RULES — READ CAREFULLY, THEY ARE MACHINE-CHECKED:
+1. A change described only in <notes> is NOT a change. If you changed a
+   file, its COMPLETE new content MUST appear in a file block above, or the
+   change is discarded and this attempt counts as FAILED.
+2. If nothing in the evidence justifies a change, you MUST emit an explicit
+   <no-change>...</no-change> block with the reason, and no file blocks.
+   Saying "no change needed" only in <notes> is NOT enough — without the
+   marker the attempt counts as FAILED.
+3. File blocks and a <no-change> marker in the same reply contradict each
+   other and the attempt counts as FAILED.
+"""
+
+_SELECTOR_OUTPUT_FORMAT = """
+
+# OUTPUT FORMAT — YOU MUST FOLLOW THIS EXACTLY
+Reply with a ```choice block containing only the chosen candidate id, and a
+<why>...</why> block explaining the decision in one paragraph.
+"""
+
+_REFLECTOR_OUTPUT_FORMAT = """
+
+# EVIDENCE NOTE
+context/harness/ is the adopted harness; context/base_harness/ is the
+unmodified base harness; context/pool/<label>/ holds each candidate harness
+plus its rollout. Compare them before writing.
+
+# OUTPUT FORMAT — YOU MUST FOLLOW THIS EXACTLY
+Reply with an <edits>...</edits> block (one row per changed component, with
+its observed effect and a verdict), then an <experience>...</experience>
+block with exactly these lines: `cue: ...`, `lesson: ...`, `evidence: ...`,
+`confidence: ...`. If no lesson is supported by an observed contrast, reply
+with the single line NO_ENTRY instead of inventing one.
+"""
+
+
+# ---------------------------------------------------------------------------
+# 客户端协议与 HTTP 实现
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class TaskState:
+    task_id: str
+    status: str
+    result_text: str = ""
+    error_text: str = ""
+
+
+class TaskClient(Protocol):
+    """控制面客户端协议：HTTP 实现见 HttpTaskClient，测试用假实现。"""
+
+    def upload(self, files: dict[str, bytes]) -> str:
+        """上传一包文件（相对路径 -> 内容），返回 upload_id。"""
+        ...
+
+    def create_task(self, *, project_id: str, prompt: str, upload_id: str | None = None,
+                    idempotency_key: str | None = None, model: str | None = None,
+                    effort: str | None = None, approval: str | None = None,
+                    environment_id: str | None = None, executor: str | None = None,
+                    no_time_limit: bool = False) -> str:
+        """建任务，返回任务 id；同幂等键返回已有任务 id，不建重复。"""
+        ...
+
+    def get_task(self, task_id: str) -> TaskState:
+        ...
+
+    def get_trace_text(self, task_id: str) -> str:
+        """轨迹渲染文本；读不到时返回空串（调用方用最终回复兜底）。"""
+        ...
+
+    def get_cost_usd(self, task_id: str) -> float:
+        """有价成本 USD；读不到时返回 0.0，不炸轮次。"""
+        ...
+
+    def cancel_task(self, task_id: str) -> None:
+        ...
+
+
+def build_multipart(files: dict[str, bytes]) -> tuple[bytes, str]:
+    """按服务端 cloud/uploads.py 的解析器能读懂的格式编码 multipart。
+
+    paths 字段是 JSON 字符串数组，与 file part 一一对应；part 名用
+    file（服务端只按顺序取 files，不认 part 名）。编码结果可用
+    cloud.uploads.parse_multipart + parse_paths_field 原样解回，
+    见 test_harness2_tasks.py 的往返测试。
+    """
+    boundary = "----harness2-" + uuid.uuid4().hex
+    paths = sorted(files)
+    parts: list[bytes] = []
+    parts.append(
+        b"--" + boundary.encode() + b"\r\n"
+        b'Content-Disposition: form-data; name="paths"\r\n\r\n'
+        + json.dumps(paths).encode() + b"\r\n"
+    )
+    for path in paths:
+        parts.append(
+            b"--" + boundary.encode() + b"\r\n"
+            + f'Content-Disposition: form-data; name="file"; filename="{path}"\r\n'.encode()
+            + b"Content-Type: application/octet-stream\r\n\r\n"
+            + files[path] + b"\r\n"
+        )
+    parts.append(b"--" + boundary.encode() + b"--\r\n")
+    return b"".join(parts), f"multipart/form-data; boundary={boundary}"
+
+
+class HttpTaskClient:
+    """基于标准库 urllib 的控制面客户端（零第三方依赖，与项目风格一致）。
+
+    瞬断重试（2026-10-29 真实首跑教训）：VM 到公网控制面的连接会
+    偶发中途断开（RemoteDisconnected / IncompleteRead），一次断开
+    就让整轮提前收尾，尽管服务端任务还在跑。_request 对传输错误与
+    502 / 503 / 429 最多重试 3 次（退避 1s、3s），其它 HTTP 错误
+    （401 / 403 / 400 等服务端裁决）不重试。重试安全的前提是调用
+    方都带 idempotency_key（本层所有建任务调用都带）与“上传包
+    未绑定即孤儿包”的语义：重试建任务时服务端返回已有任务，不
+    重复建；重试上传时最多多出一个 24 小时后过期的孤儿包。
+    """
+
+    def __init__(self, base_url: str, api_key: str, *, timeout: int = 60,
+                 max_attempts: int = 3):
+        self.base_url = base_url.rstrip("/")
+        self.api_key = api_key
+        self.timeout = timeout
+        self.max_attempts = max(1, max_attempts)
+
+    def _request(self, method: str, path: str, body: bytes | None = None,
+                 content_type: str | None = None) -> dict:
+        req = urllib.request.Request(
+            self.base_url + path, data=body, method=method)
+        req.add_header("X-API-Key", self.api_key)
+        if content_type:
+            req.add_header("Content-Type", content_type)
+        elif body is not None:
+            req.add_header("Content-Type", "application/json")
+        last_exc: Exception | None = None
+        for attempt in range(1, self.max_attempts + 1):
+            try:
+                with urllib.request.urlopen(req, timeout=self.timeout) as resp:
+                    raw = resp.read()
+            except urllib.error.HTTPError as exc:
+                if exc.code in (502, 503, 429) and attempt < self.max_attempts:
+                    last_exc = exc
+                    time.sleep(1 if attempt == 1 else 3)
+                    continue
+                detail = exc.read().decode("utf-8", errors="replace")[:500]
+                raise RuntimeError(
+                    f"{method} {path} -> HTTP {exc.code}: {detail}") from exc
+            except (urllib.error.URLError, ConnectionError, TimeoutError,
+                    OSError) as exc:
+                # RemoteDisconnected / IncompleteRead 都落在这里（前者是
+                # ConnectionResetError / http.client 子类，后者是
+                # http.client.HTTPException，而 HTTPException 不是
+                # OSError——见下方 except。
+                if attempt < self.max_attempts:
+                    last_exc = exc
+                    time.sleep(1 if attempt == 1 else 3)
+                    continue
+                raise RuntimeError(
+                    f"{method} {path} -> transport error: {exc}") from exc
+            except Exception as exc:  # http.client.HTTPException 等非 OSError
+                if exc.__class__.__name__ in (
+                        "RemoteDisconnected", "IncompleteRead", "BadStatusLine",
+                        "HTTPException") and attempt < self.max_attempts:
+                    last_exc = exc
+                    time.sleep(1 if attempt == 1 else 3)
+                    continue
+                raise
+            break
+        else:
+            raise RuntimeError(
+                f"{method} {path} -> transport error: {last_exc}")
+        try:
+            obj = json.loads(raw.decode("utf-8"))
+        except ValueError as exc:
+            raise RuntimeError(f"{method} {path} -> non-JSON response") from exc
+        if not isinstance(obj, dict):
+            raise RuntimeError(f"{method} {path} -> unexpected response shape")
+        return obj
+
+    def upload(self, files: dict[str, bytes]) -> str:
+        if not files:
+            raise ValueError("upload requires at least one file")
+        body, content_type = build_multipart(files)
+        obj = self._request("POST", "/uploads", body=body, content_type=content_type)
+        upload_id = obj.get("upload_id")
+        if not upload_id:
+            raise RuntimeError(f"/uploads did not return an upload_id: {obj!r}"[:300])
+        return str(upload_id)
+
+    def create_task(self, *, project_id: str, prompt: str, upload_id: str | None = None,
+                    idempotency_key: str | None = None, model: str | None = None,
+                    effort: str | None = None, approval: str | None = None,
+                    environment_id: str | None = None, executor: str | None = None,
+                    no_time_limit: bool = False) -> str:
+        payload: dict = {"project_id": project_id, "prompt": prompt}
+        if upload_id is not None:
+            payload["upload_id"] = upload_id
+        if idempotency_key is not None:
+            payload["idempotency_key"] = idempotency_key
+        if model is not None:
+            payload["model"] = model
+        if effort is not None:
+            payload["effort"] = effort
+        if approval is not None:
+            payload["approval"] = approval
+        if environment_id is not None:
+            payload["environment_id"] = environment_id
+        if executor is not None:
+            payload["executor"] = executor
+        if no_time_limit:
+            payload["no_time_limit"] = True
+        obj = self._request("POST", "/tasks", body=json.dumps(payload).encode())
+        task_id = obj.get("id")
+        if not task_id:
+            raise RuntimeError(f"/tasks did not return an id: {obj!r}"[:300])
+        return str(task_id)
+
+    def get_task(self, task_id: str) -> TaskState:
+        obj = self._request("GET", f"/tasks/{task_id}")
+        return TaskState(
+            task_id=task_id,
+            status=str(obj.get("status") or ""),
+            result_text=str(obj.get("result_text") or ""),
+            error_text=str(obj.get("error_text") or ""),
+        )
+
+    def get_trace_text(self, task_id: str) -> str:
+        try:
+            obj = self._request("GET", f"/tasks/{task_id}/trace")
+        except Exception:
+            return ""
+        entries = obj.get("entries") or []
+        lines: list[str] = []
+        for entry in entries:
+            if not isinstance(entry, dict):
+                continue
+            kind = entry.get("kind") or entry.get("type") or "entry"
+            text = entry.get("text") or entry.get("message") or ""
+            if not text:
+                text = json.dumps(entry, ensure_ascii=False)[:1000]
+            lines.append(f"[{kind}] {text}")
+        rendered = "\n".join(lines).strip()
+        return rendered[:TRACE_MAX_CHARS] if rendered else ""
+
+    def get_cost_usd(self, task_id: str) -> float:
+        try:
+            obj = self._request("GET", f"/tasks/{task_id}/cost")
+        except Exception:
+            return 0.0
+        try:
+            return float(obj.get("usd_micros") or 0) / 1_000_000
+        except (TypeError, ValueError):
+            return 0.0
+
+    def cancel_task(self, task_id: str) -> None:
+        try:
+            self._request("POST", f"/tasks/{task_id}/cancel", body=b"{}")
+        except Exception:
+            pass  # 终态任务取消会 400，本来就不用再取消；其它错误也不炸轮次
+
+
+@dataclass
+class CompletedTask:
+    task_id: str
+    state: TaskState
+    cost_usd: float = 0.0
+    trace_text: str = ""
+
+
+def run_task(
+    client: TaskClient,
+    *,
+    project_id: str,
+    prompt: str,
+    files: dict[str, bytes] | None = None,
+    idempotency_key: str | None = None,
+    model: str | None = None,
+    effort: str | None = None,
+    approval: str | None = None,
+    environment_id: str | None = None,
+    executor: str | None = None,
+    no_time_limit: bool = False,
+    timeout_seconds: int = DEFAULT_TASK_TIMEOUT_SECONDS,
+    poll_seconds: int = DEFAULT_POLL_SECONDS,
+) -> CompletedTask:
+    """建任务、轮询到终态、读成本与轨迹。超时先取消再抛 TimeoutError。"""
+    upload_id = client.upload(files) if files else None
+    task_id = client.create_task(
+        project_id=project_id, prompt=prompt, upload_id=upload_id,
+        idempotency_key=idempotency_key, model=model, effort=effort,
+        approval=approval, environment_id=environment_id, executor=executor,
+        no_time_limit=no_time_limit)
+    deadline = time.time() + timeout_seconds
+    state = client.get_task(task_id)
+    while state.status not in TERMINAL_STATUSES:
+        if time.time() >= deadline:
+            client.cancel_task(task_id)
+            raise TimeoutError(
+                f"task {task_id} did not reach a terminal status within "
+                f"{timeout_seconds}s (last status: {state.status})")
+        time.sleep(poll_seconds)
+        state = client.get_task(task_id)
+    # 拆信封的唯一位置：run_task 是探针、编辑、选择、反思四路的
+    # 共同 choke point，在这里拆一次，全路受益；各 TaskClient 的
+    # get_task 保持原样返回服务端原文。
+    state = TaskState(
+        task_id=state.task_id, status=state.status,
+        result_text=unwrap_result_text(state.result_text),
+        error_text=state.error_text,
+    )
+    return CompletedTask(
+        task_id=task_id, state=state,
+        cost_usd=client.get_cost_usd(task_id),
+        trace_text=client.get_trace_text(task_id),
+    )
+
+
+def _flatten_dir(root: Path, prefix: str = "") -> dict[str, bytes]:
+    """把一个目录展平成 {相对路径: 内容}；符号链接与 __pycache__ 跳过。"""
+    out: dict[str, bytes] = {}
+    root = Path(root)
+    if not root.is_dir():
+        return out
+    for p in sorted(root.rglob("*")):
+        if not p.is_file() or p.is_symlink() or "__pycache__" in p.parts:
+            continue
+        rel = str(p.relative_to(root))
+        out[(prefix + rel) if prefix else rel] = p.read_bytes()
+    return out
+
+
+def _tree_to_files(tree: Tree) -> dict[str, bytes]:
+    return {rel: content.encode("utf-8") for rel, content in tree.items()}
+
+
+# ---------------------------------------------------------------------------
+# 三个协议的真实任务实现
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class CodexProbeAdapter:
+    """探针 adapter：每个 label 跑一个真实 codex 任务。
+
+    source_files：原任务需要的源文件（如 ledger.csv），会随 harness
+    一起 seed 进 worktree；与 harness 路径冲突时 harness 优先（探针
+    的意义就是让 harness 覆盖仓库默认值）。调用方应在 source_files
+    里避免与 harness 路径重名——重名时源文件会被静默覆盖，本层
+    不为此报错（与 seed“用户文件优先”的语义一致）。
+    """
+
+    client: TaskClient
+    project_id: str
+    task_id: str
+    prompt: str
+    base: Tree
+    run_tag: str
+    source_files: dict[str, bytes] = field(default_factory=dict)
+    environment_id: str | None = None
+    executor: str | None = None
+    model: str | None = None
+    effort: str | None = None
+    approval: str | None = None
+    no_time_limit: bool = False
+    timeout_seconds: int = DEFAULT_TASK_TIMEOUT_SECONDS
+    domain: str = "general"
+
+    def base_harness(self) -> Tree:
+        return dict(self.base)
+
+    def execute(self, label: str, harness: Tree) -> Rollout:
+        files = dict(self.source_files)
+        files.update(_tree_to_files(harness))
+        try:
+            done = run_task(
+                self.client, project_id=self.project_id,
+                prompt=self.prompt + _PROBE_DELIVERABLE_FOOTER,
+                files=files,
+                idempotency_key=f"h2-{self.run_tag}-probe-{label}",
+                model=self.model, effort=self.effort, approval=self.approval,
+                environment_id=self.environment_id, executor=self.executor,
+                no_time_limit=self.no_time_limit,
+                timeout_seconds=self.timeout_seconds)
+        except Exception as exc:
+            return Rollout(label=label, ok=False, delivered=False,
+                           error=f"{type(exc).__name__}: {exc}"[:400])
+        state = done.state
+        delivered = state.status == "completed" and bool(state.result_text.strip())
+        return Rollout(
+            label=label, ok=state.status == "completed", delivered=delivered,
+            deliverable_text=state.result_text,
+            trajectory=done.trace_text or state.result_text,
+            cost_usd=done.cost_usd,
+            error="" if state.status == "completed" else (
+                state.error_text or f"task ended as {state.status}")[:400],
+        )
+
+
+@dataclass
+class CodexEditor:
+    """编辑器：一个 codex 任务，证据 seed 进 worktree，改动以文件块回读。"""
+
+    client: TaskClient
+    project_id: str
+    run_tag: str
+    environment_id: str | None = None
+    executor: str | None = None
+    model: str | None = None
+    effort: str | None = None
+    approval: str | None = None
+    no_time_limit: bool = False
+    timeout_seconds: int = DEFAULT_TASK_TIMEOUT_SECONDS
+
+    def edit(self, mission: str, context_dir: Path, label: str) -> EditResult:
+        """编辑任务：格式错误给一次带反馈的纠正机会，再错才如实回落。
+
+        背景（2026-10-29 两轮真实轮次）：编辑在 <notes> 里声称改
+        动、却不发文件块，旧契约把 notes-only 当“无需改动”采纳，
+        改动无声丢失；且方法层的阶段循环里编辑异常是直接 break（不
+        重试），所以纠正必须发生在本层。格式重试用独立幂等键
+        （…-edit-{label}.fmt2），不复用首个任务、不重复计费到同
+        一任务；任务本身失败（非 completed）仍直接抛错、不重试。
+        """
+        files = _flatten_dir(context_dir, prefix="context/")
+        files["context/MISSION.md"] = mission.encode("utf-8")
+        last_err = ""
+        total_cost = 0.0
+        for attempt in (1, 2):
+            prompt = mission + _EDITOR_OUTPUT_FORMAT
+            if last_err:
+                prompt += (
+                    "\n\n# YOUR PREVIOUS REPLY WAS REJECTED\n\n"
+                    f"{last_err}\n\nFix the output format and reply again: "
+                    "changed files MUST appear as === HARNESS FILE === blocks "
+                    "with their COMPLETE new content, or, if truly no change "
+                    "is justified, emit <no-change>...</no-change>.")
+            done = run_task(
+                self.client, project_id=self.project_id,
+                prompt=prompt,
+                files=files,
+                idempotency_key=f"h2-{self.run_tag}-edit-{label}"
+                + ("" if attempt == 1 else ".fmt2"),
+                model=self.model, effort=self.effort, approval=self.approval,
+                environment_id=self.environment_id, executor=self.executor,
+                no_time_limit=self.no_time_limit,
+                timeout_seconds=self.timeout_seconds)
+            state = done.state
+            if state.status != "completed":
+                raise RuntimeError(
+                    f"editor task {done.task_id} ended as {state.status}: "
+                    f"{state.error_text}"[:400])
+            total_cost += done.cost_usd
+            text = state.result_text or ""
+            try:
+                changed = parse_harness_files(text)
+                notes = parse_notes(text)
+                no_change = parse_no_change(text)
+            except ValueError as exc:
+                last_err = f"output-format error: {exc}"
+                continue
+            if changed and no_change:
+                last_err = ("output-format error: reply contains both harness "
+                            "file blocks and a <no-change> marker")
+                continue
+            if not changed and not no_change:
+                last_err = ("output-format error: reply contains neither "
+                            "harness file blocks nor a <no-change> marker; "
+                            "a change described only in <notes> is discarded")
+                continue
+            base_tree = read_tree(Path(context_dir) / "harness")
+            merged = dict(base_tree)
+            merged.update(changed)
+            return EditResult(tree=merged, notes=notes, cost_usd=total_cost)
+        raise ValueError(
+            f"editor reply failed output-format validation twice: {last_err}")
+
+
+@dataclass
+class CodexSelector:
+    """选择器：一个 codex 任务，候选产物 seed 进 worktree，只解析最终回复。"""
+
+    client: TaskClient
+    project_id: str
+    run_tag: str
+    environment_id: str | None = None
+    executor: str | None = None
+    model: str | None = None
+    effort: str | None = None
+    approval: str | None = None
+    no_time_limit: bool = False
+    timeout_seconds: int = DEFAULT_TASK_TIMEOUT_SECONDS
+
+    def select(self, mission: str, context_dir: Path) -> tuple[str, str, float]:
+        files = _flatten_dir(context_dir, prefix="context/")
+        files["context/MISSION.md"] = mission.encode("utf-8")
+        done = run_task(
+            self.client, project_id=self.project_id,
+            prompt=mission + _SELECTOR_OUTPUT_FORMAT,
+            files=files,
+            idempotency_key=f"h2-{self.run_tag}-select-{Path(context_dir).name}",
+            model=self.model, effort=self.effort, approval=self.approval,
+            environment_id=self.environment_id, executor=self.executor,
+            no_time_limit=self.no_time_limit,
+            timeout_seconds=self.timeout_seconds)
+        state = done.state
+        if state.status != "completed":
+            raise RuntimeError(
+                f"selector task {done.task_id} ended as {state.status}: "
+                f"{state.error_text}"[:400])
+        choice = parse_choice(state.result_text or "")
+        if not choice:
+            raise ValueError("selector task returned no ```choice block")
+        return choice, parse_why(state.result_text or ""), done.cost_usd
+
+
+@dataclass
+class CodexReflector:
+    """反思器：一个 codex 任务，解析 <experience>；三项不全返回 {}。"""
+
+    client: TaskClient
+    project_id: str
+    run_tag: str
+    environment_id: str | None = None
+    executor: str | None = None
+    model: str | None = None
+    effort: str | None = None
+    approval: str | None = None
+    no_time_limit: bool = False
+    timeout_seconds: int = DEFAULT_TASK_TIMEOUT_SECONDS
+
+    def reflect(self, mission: str, context_dir: Path) -> dict:
+        files = _flatten_dir(context_dir, prefix="context/")
+        files["context/MISSION.md"] = mission.encode("utf-8")
+        done = run_task(
+            self.client, project_id=self.project_id,
+            prompt=mission + _REFLECTOR_OUTPUT_FORMAT,
+            files=files,
+            idempotency_key=f"h2-{self.run_tag}-reflect-{Path(context_dir).name}",
+            model=self.model, effort=self.effort, approval=self.approval,
+            environment_id=self.environment_id, executor=self.executor,
+            no_time_limit=self.no_time_limit,
+            timeout_seconds=self.timeout_seconds)
+        state = done.state
+        if state.status != "completed":
+            return {}
+        return parse_experience(state.result_text or "")
+
+
+def run_improvement_via_tasks(
+    client: TaskClient,
+    *,
+    project_id: str,
+    task_id: str,
+    prompt: str,
+    base_harness: Tree,
+    run_tag: str,
+    work_dir: Path,
+    domain: str = "general",
+    source_files: dict[str, bytes] | None = None,
+    k: int = 1,
+    mode: str = "parallel",
+    library_root: Path | None = None,
+    budget_usd: float | None = None,
+    model: str | None = None,
+    effort: str | None = None,
+    approval: str | None = None,
+    environment_id: str | None = None,
+    executor: str | None = None,
+    no_time_limit: bool = False,
+    timeout_seconds: int = DEFAULT_TASK_TIMEOUT_SECONDS,
+):
+    """一站式：用真实 codex 任务跑一轮完整改进，返回 ImprovementResult。
+
+    library_root 为空时不建经验库（只跑改进、不入库）；非空时按
+    domain 建库。auto_apply 恒为 False——采纳的树只入库待评审，
+    不自动成为后续任务的 base（见 docs/harness2.md 安全一节）。
+    """
+    from .harness2 import Library
+
+    adapter = CodexProbeAdapter(
+        client=client, project_id=project_id, task_id=task_id, prompt=prompt,
+        base=base_harness, run_tag=run_tag,
+        source_files=source_files or {}, environment_id=environment_id,
+        executor=executor, model=model, effort=effort, approval=approval,
+        no_time_limit=no_time_limit, timeout_seconds=timeout_seconds,
+        domain=domain)
+    shared = dict(environment_id=environment_id, executor=executor, model=model,
+                  effort=effort, approval=approval, no_time_limit=no_time_limit,
+                  timeout_seconds=timeout_seconds)
+    editor = CodexEditor(client=client, project_id=project_id, run_tag=run_tag, **shared)
+    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
+    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)
+
+
+__all__ = [
+    "TERMINAL_STATUSES", "DEFAULT_TASK_TIMEOUT_SECONDS", "DEFAULT_POLL_SECONDS",
+    "TRACE_MAX_CHARS", "TaskState", "TaskClient", "build_multipart",
+    "HttpTaskClient", "CompletedTask", "run_task",
+    "CodexProbeAdapter", "CodexEditor", "CodexSelector", "CodexReflector",
+    "run_improvement_via_tasks",
+]

cloud/tests/test_harness2.py

@@ -0,0 +1,335 @@
+"""cloud/harness2.py 单测：递归改进循环。
+
+不跑真实模型、不联网：执行器在进程内跑 harness 树里的脚本，
+编辑器/选择器/反思器都是确定性假实现（上游 harness2-demo 的同款
+做法：base 脚本只加正数，编辑器去掉正数过滤，正确总数是带符号和）。
+
+账本：100.25, -20.10, 5.00 —— 带符号和 85.15，只加正数是 105.25。
+"""
+
+import csv
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+
+try:
+    from cloud import harness2
+    from cloud.harness2 import (
+        Library, Rollout, EditResult, Budget, build_context, validate,
+        read_tree, write_tree, is_allowed, normalize_rel, autotrigger,
+        skill_names, diff_summary, diversity, improve_task, select_final,
+        assign_ids, has_experience,
+    )
+except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud import harness2
+    from cloud.harness2 import (
+        Library, Rollout, EditResult, Budget, build_context, validate,
+        read_tree, write_tree, is_allowed, normalize_rel, autotrigger,
+        skill_names, diff_summary, diversity, improve_task, select_final,
+        assign_ids, has_experience,
+    )
+
+BASE_SCRIPT = """import csv
+import sys
+from decimal import Decimal
+with open(sys.argv[1]) as f:
+    amounts = [Decimal(r['amount']) for r in csv.DictReader(f)]
+print(f"Total: {sum((a for a in amounts if a > 0), Decimal(0)):.2f}")
+"""
+
+BASE_TREE = {
+    "AGENTS.md": "Use scripts/summarize.py to total the ledger.\n",
+    "scripts/summarize.py": BASE_SCRIPT,
+}
+
+
+class LedgerAdapter:
+    """任务：对账本求和。执行 = 真跑 harness 树里的脚本（无模型）。"""
+
+    task_id = "ledger-1"
+    domain = "accounting"
+    prompt = "Sum every signed amount in ledger.csv. Return Total: with two decimals."
+
+    def __init__(self, root):
+        self.root = Path(root)
+        self.executions = 0
+        self.ledger = self.root / "ledger.csv"
+        self.ledger.write_text("amount\n100.25\n-20.10\n5.00\n", encoding="utf-8")
+
+    def base_harness(self):
+        return dict(BASE_TREE)
+
+    def execute(self, label, harness):
+        self.executions += 1
+        work = self.root / "runs" / label
+        work.mkdir(parents=True, exist_ok=True)
+        write_tree(harness, work / "harness")
+        proc = subprocess.run(
+            [sys.executable, str(work / "harness" / "scripts" / "summarize.py"),
+             str(self.ledger)],
+            capture_output=True, text=True, timeout=10,
+        )
+        out = proc.stdout.strip()
+        return Rollout(
+            label=label, ok=proc.returncode == 0, delivered=out.startswith("Total:"),
+            deliverable_text=out, trajectory="[ACTION] Run scripts/summarize.py\n"
+            "[OBSERVATION]\n" + out, cost_usd=0.01,
+            error="" if proc.returncode == 0 else proc.stderr[-200:],
+        )
+
+    def validate_tree(self, harness):
+        if "scripts/summarize.py" not in harness:
+            raise ValueError("The local executor requires scripts/summarize.py")
+
+
+class FixEditor:
+    """确定性编辑器：去掉正数过滤；同时在 context/harness/ 里真实改文件，
+    模拟 codex 任务编辑工作目录的方式，再把改完的树读回。"""
+
+    def edit(self, mission, context_dir, label):
+        harness_dir = Path(context_dir) / "harness"
+        script = harness_dir / "scripts" / "summarize.py"
+        script.write_text(script.read_text().replace(" if a > 0", ""), encoding="utf-8")
+        return EditResult(tree=read_tree(harness_dir),
+                          notes="Keep signed amounts: the base total omitted credits.")
+
+
+class BadEditor:
+    """永远交回非法树：AGENTS.md 为空，校验必拒，最终必须回落到 base。"""
+
+    def edit(self, mission, context_dir, label):
+        return EditResult(tree={"AGENTS.md": ""}, notes="bad proposal")
+
+
+class PickSignedSelector:
+    def select(self, mission, context_dir):
+        for path in sorted((Path(context_dir) / "candidates").glob("*.md")):
+            if "Total: 85.15" in path.read_text():
+                return path.stem, "The signed total matches the source ledger.", 0.0
+        raise ValueError("no candidate with the signed total")
+
+
+class PickFirstSelector:
+    def select(self, mission, context_dir):
+        paths = sorted((Path(context_dir) / "candidates").glob("*.md"))
+        return paths[0].stem, "first delivered candidate", 0.0
+
+
+class RaisingSelector:
+    def select(self, mission, context_dir):
+        raise RuntimeError("selector session failed")
+
+
+class LedgerReflector:
+    def reflect(self, mission, context_dir):
+        root = Path(context_dir)
+        changed = (root / "harness" / "scripts" / "summarize.py").read_text() != (
+            root / "base_harness" / "scripts" / "summarize.py").read_text()
+        if not changed:
+            return {}
+        return {
+            "cue": "When a ledger includes credits and debits",
+            "lesson": "Include both signs when summing in scripts/summarize.py.",
+            "evidence": "Removing the positive-only filter changed the probe total "
+                        "to include the credits omitted by the base trajectory.",
+            "confidence": "high for this observed arithmetic contrast",
+            "edits": "scripts/summarize.py: dropped the positive-only filter; observed: "
+                     "probe total included the omitted credits; verdict: helped",
+        }
+
+
+class TestTree(unittest.TestCase):
+    def test_validate_accepts_base(self):
+        census = validate(dict(BASE_TREE))
+        self.assertEqual(census["scripts"], 1)
+        self.assertEqual(census["n_files"], 2)
+
+    def test_validate_rejects_skill_without_description(self):
+        tree = dict(BASE_TREE)
+        tree[".agents/skills/check/SKILL.md"] = "Do the check.\n"
+        tree["AGENTS.md"] += "- When checking, read .agents/skills/check/SKILL.md.\n"
+        with self.assertRaises(ValueError) as cm:
+            validate(tree)
+        self.assertIn("description", str(cm.exception))
+
+    def test_validate_rejects_untriggered_skill(self):
+        tree = dict(BASE_TREE)
+        tree[".agents/skills/check/SKILL.md"] = "---\ndescription: check totals\n---\nDo it.\n"
+        with self.assertRaises(ValueError) as cm:
+            validate(tree)
+        self.assertIn("never named", str(cm.exception))
+
+    def test_validate_rejects_bad_hooks_and_bad_script(self):
+        tree = dict(BASE_TREE)
+        tree[".codex/hooks.json"] = '{"hooks": {"Nope": []}}'
+        with self.assertRaises(ValueError):
+            validate(tree)
+        tree = dict(BASE_TREE)
+        tree["scripts/summarize.py"] = "def broken(:\n"
+        with self.assertRaises(ValueError):
+            validate(tree)
+        tree = dict(BASE_TREE)
+        tree["AGENTS.md"] = "x" * 8001
+        with self.assertRaises(ValueError):
+            validate(tree)
+
+    def test_validate_rejects_missing_script_ref(self):
+        tree = dict(BASE_TREE)
+        tree["AGENTS.md"] += "Run python3 scripts/nope.py first.\n"
+        with self.assertRaises(ValueError):
+            validate(tree)
+
+    def test_autotrigger_repairs_and_respects_cap(self):
+        tree = dict(BASE_TREE)
+        tree[".agents/skills/check/SKILL.md"] = "---\ndescription: check totals\n---\nDo it.\n"
+        fixed = autotrigger(tree)
+        self.assertEqual(fixed, ["check"])
+        self.assertIn("check", tree["AGENTS.md"])
+        validate(tree)  # 补完触发行后应能通过
+        capped = dict(BASE_TREE)
+        capped["AGENTS.md"] = "x" * 7995
+        capped[".agents/skills/check/SKILL.md"] = "---\ndescription: check totals\n---\nDo it.\n"
+        self.assertEqual(autotrigger(capped), [])  # 超上限时跳过，不硬塞
+
+    def test_paths_and_roundtrip(self):
+        with self.assertRaises(ValueError):
+            normalize_rel("../escape.py")
+        self.assertFalse(is_allowed("/etc/passwd"))
+        self.assertFalse(is_allowed("evil.py"))
+        self.assertTrue(is_allowed(".agents/skills/a/SKILL.md"))
+        with tempfile.TemporaryDirectory() as tmp:
+            dest = Path(tmp) / "tree"
+            write_tree(dict(BASE_TREE), dest)
+            self.assertEqual(read_tree(dest), BASE_TREE)
+            rejected = []
+            (dest / "evil.py").write_text("x", encoding="utf-8")
+            self.assertEqual(read_tree(dest, rejected=rejected), BASE_TREE)
+            self.assertEqual(rejected, ["evil.py"])
+
+    def test_diff_and_diversity(self):
+        changed = dict(BASE_TREE)
+        changed["scripts/summarize.py"] = "changed"
+        d = diff_summary(BASE_TREE, changed)
+        self.assertEqual(d["modified"], ["scripts/summarize.py"])
+        self.assertEqual(d["added"], [])
+        div = diversity({"a": BASE_TREE, "b": changed})
+        self.assertEqual(div["n_distinct"], 2)
+        self.assertLess(div["mean_line_jaccard"], 1.0)
+
+
+class TestImprove(unittest.TestCase):
+    def _run(self, mode, editor=None, selector=None, budget=None, k=1):
+        tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(tmp.cleanup)
+        root = Path(tmp.name)
+        adapter = LedgerAdapter(root)
+        library = Library(root / "library", "accounting")
+        result = improve_task(
+            adapter, editor or FixEditor(),
+            selector=selector or PickSignedSelector(),
+            reflector=LedgerReflector(), work_dir=root / "work",
+            k=k, mode=mode, library=library, budget_usd=budget)
+        return result, adapter, library, root
+
+    def test_parallel_k1_improves_and_enters_library(self):
+        result, adapter, library, root = self._run("parallel")
+        self.assertTrue(result.ok)
+        self.assertEqual(result.selection.chosen_label, "agg1")
+        chosen = result.mode_result.rollouts["agg1"]
+        self.assertEqual(chosen.deliverable_text, "Total: 85.15")
+        self.assertEqual(adapter.executions, 4)  # base + inv1 + dom1 + agg1
+        self.assertIsNotNone(result.library_entry)
+        self.assertIn("ledger-1", library.entries())
+        index = (root / "library" / "accounting" / "INDEX.md").read_text()
+        self.assertIn("credits and debits", index)
+        self.assertIn("85.15", result.selection.why + chosen.deliverable_text)
+
+    def test_sequential_k1_improves(self):
+        result, adapter, library, root = self._run("sequential")
+        self.assertTrue(result.ok)
+        self.assertEqual(result.selection.chosen_label, "c1")
+        self.assertEqual(result.mode_result.rollouts["c1"].deliverable_text, "Total: 85.15")
+        self.assertEqual(adapter.executions, 4)  # base + inv1 + dom1 + c1
+
+    def test_invalid_proposal_falls_back_to_base(self):
+        result, adapter, library, root = self._run(
+            "parallel", editor=BadEditor(), selector=PickFirstSelector())
+        self.assertTrue(result.ok)  # 回落的 base 产物仍可交付
+        self.assertEqual(set(result.mode_result.fell_back), {"inv1", "dom1", "agg1"})
+        self.assertEqual(result.mode_result.trees["agg1"], BASE_TREE)
+        self.assertIsNone(result.library_entry)  # 树没变，反思无经验，不入库
+
+    def test_budget_exceeded_blocks_library_but_still_delivers(self):
+        result, adapter, library, root = self._run("parallel", budget=0.03)
+        self.assertTrue(result.ok)
+        self.assertTrue(result.budget_exceeded)
+        self.assertIsNone(result.library_entry)
+
+    def test_k_validation(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            adapter = LedgerAdapter(Path(tmp))
+            with self.assertRaises(ValueError):
+                improve_task(adapter, FixEditor(), work_dir=Path(tmp) / "w", k=0)
+            with self.assertRaises(ValueError):
+                improve_task(adapter, FixEditor(), work_dir=Path(tmp) / "w", k=4)
+            with self.assertRaises(ValueError):
+                improve_task(adapter, FixEditor(), work_dir=Path(tmp) / "w", k=1,
+                             mode="sideways")
+
+    def test_trivial_k1_without_selector(self):
+        tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(tmp.cleanup)
+        root = Path(tmp.name)
+        adapter = LedgerAdapter(root)
+        result = improve_task(adapter, FixEditor(), selector=None,
+                              reflector=LedgerReflector(), work_dir=root / "work",
+                              k=1, mode="parallel")
+        self.assertTrue(result.ok)
+        self.assertTrue(result.selection.trivial)
+
+
+class TestSelection(unittest.TestCase):
+    def test_drops_undelivered_and_falls_back_on_selector_error(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            adapter = LedgerAdapter(Path(tmp))
+            rollouts = {
+                "agg1": Rollout(label="agg1", ok=True, delivered=True,
+                                deliverable_text="Total: 1.00"),
+                "agg2": Rollout(label="agg2", ok=False, delivered=False),
+            }
+            # k=2 的终版是 agg1/agg2；agg2 未交付应被丢掉，agg1 是唯一候选
+            sel = select_final("ledger-1", 2, "parallel", rollouts, adapter=adapter,
+                               selector=RaisingSelector(), work_dir=Path(tmp) / "work",
+                               base_tree=dict(BASE_TREE))
+            self.assertTrue(sel.ok)
+            self.assertTrue(sel.fallback)
+            self.assertEqual(sel.chosen_label, "agg1")
+            self.assertIn("agg2", sel.dropped)
+
+    def test_assign_ids_is_stable(self):
+        a = assign_ids(["agg2", "agg1"], task_id="t", k=1, mode="parallel")
+        b = assign_ids(["agg1", "agg2"], task_id="t", k=1, mode="parallel")
+        self.assertEqual(a, b)  # 输入顺序不影响匿名映射
+
+
+class TestLibrary(unittest.TestCase):
+    def test_add_requires_experience(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            lib = Library(Path(tmp) / "lib", "accounting")
+            self.assertIsNone(lib.add("t1", dict(BASE_TREE), experience={}))
+            self.assertEqual(lib.entries(), [])
+            self.assertIsNone(lib.context_dir())
+            entry = lib.add("t1", dict(BASE_TREE), why="why", label="agg1",
+                            experience={"cue": "c", "lesson": "l", "evidence": "e"})
+            self.assertIsNotNone(entry)
+            self.assertEqual(lib.entries(), ["t1"])
+            self.assertEqual(lib.tree_of("t1"), BASE_TREE)
+            self.assertIsNotNone(lib.context_dir())
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/tests/test_harness2_tasks.py

@@ -0,0 +1,491 @@
+"""cloud/harness2_tasks.py 单测：真实 codex 任务版接线。
+
+不跑真实模型、不联网：FakeTaskClient 在进程内模拟控制面——探针
+任务真跑 bundle 里的脚本，编辑任务按文件块格式回改动，选择与
+反思按最终回复格式回结果。HTTP 客户端则对本地假服务端实测。
+账本与 base 树同 test_harness2.py：带符号和 85.15，只加正数 105.25。
+"""
+
+import csv
+import json
+import subprocess
+import sys
+import tempfile
+import threading
+import unittest
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+
+try:
+    from cloud import harness2_tasks
+    from cloud.harness2 import (
+        parse_choice, parse_experience, parse_harness_files, parse_no_change,
+        parse_notes, parse_why,
+    )
+    from cloud.harness2_tasks import (
+        CodexEditor, CodexProbeAdapter, HttpTaskClient, TaskState,
+        build_multipart, run_improvement_via_tasks, run_task,
+        unwrap_result_text,
+    )
+    from cloud.uploads import parse_multipart, parse_paths_field
+except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud import harness2_tasks
+    from cloud.harness2 import (
+        parse_choice, parse_experience, parse_harness_files, parse_no_change,
+        parse_notes, parse_why,
+    )
+    from cloud.harness2_tasks import (
+        CodexEditor, CodexProbeAdapter, HttpTaskClient, TaskState,
+        build_multipart, run_improvement_via_tasks, run_task,
+        unwrap_result_text,
+    )
+    from cloud.uploads import parse_multipart, parse_paths_field
+
+BASE_SCRIPT = """import csv
+import sys
+from decimal import Decimal
+with open(sys.argv[1]) as f:
+    amounts = [Decimal(r['amount']) for r in csv.DictReader(f)]
+print(f"Total: {sum((a for a in amounts if a > 0), Decimal(0)):.2f}")
+"""
+
+BASE_TREE = {
+    "AGENTS.md": "Use scripts/summarize.py to total the ledger.\n",
+    "scripts/summarize.py": BASE_SCRIPT,
+}
+
+LEDGER = b"amount\n100.25\n-20.10\n5.00\n"
+
+
+class FakeTaskClient:
+    """进程内控制面：按 prompt 类型分派到确定性假 worker。"""
+
+    def __init__(self, editor_mode="fix", selector_mode="signed", envelope=False):
+        self.editor_mode = editor_mode
+        self.selector_mode = selector_mode
+        self.envelope = envelope
+        self.uploads = {}
+        self.tasks = {}
+        self.by_key = {}
+        self.created = 0
+        self.cancelled = []
+
+    def upload(self, files):
+        uid = f"u{len(self.uploads) + 1}"
+        self.uploads[uid] = dict(files)
+        return uid
+
+    def create_task(self, *, project_id, prompt, upload_id=None,
+                    idempotency_key=None, **kwargs):
+        if idempotency_key and idempotency_key in self.by_key:
+            return self.by_key[idempotency_key]
+        tid = f"t{len(self.tasks) + 1}"
+        self.created += 1
+        files = self.uploads.get(upload_id, {}) if upload_id else {}
+        result = self._work(prompt, files)
+        if self.envelope:
+            # 模拟控制面真实形状：result_text 是 JSON 信封，text 字段
+            # 里才是最终回复——接线层必须拆信封，否则行首锚定的文件
+            # 块解析永远看不见块（2026-10-29 真实轮次根因）。
+            result = json.dumps({
+                "task": tid, "thread_id": "th", "cwd": "/x",
+                "status": "completed", "text": result, "error": None,
+            })
+        self.tasks[tid] = TaskState(task_id=tid, status="completed",
+                                    result_text=result)
+        if idempotency_key:
+            self.by_key[idempotency_key] = tid
+        return tid
+
+    def _work(self, prompt, files):
+        if prompt.startswith("You are editing an agent harness"):
+            if self.editor_mode == "gibberish":
+                return "hello, I did stuff"
+            if self.editor_mode in ("notes-only-then-fix", "notes-only"):
+                if ("YOUR PREVIOUS REPLY WAS REJECTED" in prompt
+                        and self.editor_mode == "notes-only-then-fix"):
+                    script = files["context/harness/scripts/summarize.py"].decode()
+                    fixed = script.replace(" if a > 0", "")
+                    return (
+                        "=== HARNESS FILE: scripts/summarize.py ===\n" + fixed +
+                        "=== END HARNESS FILE ===\n"
+                        "<notes>Removed the positive-only filter.</notes>")
+                # 真实轮次原样复现：notes 里声称已修正，却不发文件块。
+                return ("<notes>Removed the positive-only filter so signed "
+                        "amounts are summed; the script now totals all amounts."
+                        "</notes>")
+            if self.editor_mode == "no-change":
+                return ("<notes>No evidence justifies a change.</notes>"
+                        "<no-change>The base trajectory shows no defect."
+                        "</no-change>")
+            script = files["context/harness/scripts/summarize.py"].decode()
+            fixed = script.replace(" if a > 0", "")
+            return (
+                "=== HARNESS FILE: scripts/summarize.py ===\n" + fixed +
+                "=== END HARNESS FILE ===\n"
+                "<notes>Keep signed amounts: the base total omitted credits.</notes>"
+            )
+        if prompt.startswith("You are choosing which"):
+            if self.selector_mode == "gibberish":
+                return "I like them all"
+            cands = sorted(p for p in files if p.startswith("context/candidates/")
+                           and p.endswith(".md"))
+            for p in cands:
+                if b"Total: 85.15" in files[p]:
+                    stem = Path(p).stem
+                    return (f"```choice\n{stem}\n```\n"
+                            "<why>The signed total matches the source ledger.</why>")
+            stem = Path(cands[0]).stem
+            return (f"```choice\n{stem}\n```\n<why>first delivered candidate</why>")
+        if prompt.startswith("You are reading the complete record"):
+            changed = (files.get("context/harness/scripts/summarize.py")
+                       != files.get("context/base_harness/scripts/summarize.py"))
+            if not changed:
+                return "NO_ENTRY"
+            return (
+                "<edits>\nscripts/summarize.py: dropped the positive-only filter; "
+                "verdict: helped\n</edits>\n"
+                "<experience>\ncue: When a ledger includes credits and debits\n"
+                "lesson: Include both signs when summing in scripts/summarize.py.\n"
+                "evidence: Removing the positive-only filter changed the probe total "
+                "to include the credits omitted by the base trajectory.\n"
+                "confidence: high\n</experience>"
+            )
+        # 探针：真跑 bundle 里的脚本对 bundle 里的账本求和。
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            for rel, data in files.items():
+                dest = root / rel
+                dest.parent.mkdir(parents=True, exist_ok=True)
+                dest.write_bytes(data)
+            proc = subprocess.run(
+                [sys.executable, str(root / "scripts" / "summarize.py"),
+                 str(root / "ledger.csv")],
+                capture_output=True, text=True, timeout=10)
+            return proc.stdout.strip()
+
+    def get_task(self, task_id):
+        return self.tasks[task_id]
+
+    def get_trace_text(self, task_id):
+        return "[ACTION] fake probe\n[OBSERVATION]\n" + self.tasks[task_id].result_text
+
+    def get_cost_usd(self, task_id):
+        return 0.02
+
+    def cancel_task(self, task_id):
+        self.cancelled.append(task_id)
+
+
+class TestParseHelpers(unittest.TestCase):
+    def test_multipart_roundtrips_through_server_parser(self):
+        files = {"context/task.md": b"do it", "AGENTS.md": b"be nice\n"}
+        body, content_type = build_multipart(files)
+        fields, parsed = parse_multipart(body, content_type)
+        paths = parse_paths_field(fields, parsed)
+        self.assertEqual(paths, sorted(files))
+        self.assertEqual(dict(parsed), files)
+
+    def test_choice_why_notes_experience(self):
+        self.assertEqual(parse_choice("```choice\nB\n```"), "B")
+        self.assertIsNone(parse_choice("no choice here"))
+        self.assertEqual(parse_why("<why>because</why>"), "because")
+        self.assertEqual(parse_notes("<notes>did x</notes>"), "did x")
+        exp = parse_experience(
+            "<experience>\ncue: c\nlesson: l\nevidence: e\nconfidence: high\n</experience>")
+        self.assertEqual((exp["cue"], exp["lesson"], exp["evidence"]), ("c", "l", "e"))
+        self.assertEqual(parse_experience("NO_ENTRY"), {})
+        self.assertEqual(parse_experience("<experience>\ncue: c\n</experience>"), {})
+
+    def test_harness_file_blocks(self):
+        text = ("=== HARNESS FILE: AGENTS.md ===\nhello\n"
+                "=== END HARNESS FILE ===\n<notes>n</notes>")
+        self.assertEqual(parse_harness_files(text), {"AGENTS.md": "hello\n"})
+        self.assertEqual(parse_harness_files("nothing here"), {})
+        with self.assertRaises(ValueError):
+            parse_harness_files("=== HARNESS FILE: /etc/passwd ===\nx\n"
+                                "=== END HARNESS FILE ===")
+        with self.assertRaises(ValueError):
+            parse_harness_files("=== HARNESS FILE: AGENTS.md ===\nunterminated\n")
+
+
+class TestFullRunViaTasks(unittest.TestCase):
+    def _run(self, client, run_tag="run1", work_sub="work"):
+        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=run_tag,
+            work_dir=root / work_sub, domain="accounting",
+            source_files={"ledger.csv": LEDGER},
+            k=1, mode="parallel", library_root=root / "library")
+        return result, client, root
+
+    def test_parallel_k1_via_real_task_shape(self):
+        client = FakeTaskClient()
+        result, _, root = self._run(client)
+        self.assertTrue(result.ok)
+        self.assertEqual(result.selection.chosen_label, "agg1")
+        self.assertEqual(result.mode_result.rollouts["agg1"].deliverable_text,
+                         "Total: 85.15")
+        # 4 探针 + 3 编辑 + 1 选择 + 1 反思 = 9 个任务
+        self.assertEqual(client.created, 9)
+        self.assertIsNotNone(result.library_entry)
+        self.assertTrue((root / "library" / "accounting" / "INDEX.md").is_file())
+
+    def test_same_run_tag_replays_without_new_tasks(self):
+        client = FakeTaskClient()
+        result1, _, _ = self._run(client, work_sub="work1")
+        self.assertTrue(result1.ok)
+        created_after_first = client.created
+        result2, _, _ = self._run(client, work_sub="work2")
+        self.assertTrue(result2.ok)
+        self.assertEqual(client.created, created_after_first)  # 幂等键命中，零新建
+
+    def test_editor_gibberish_falls_back(self):
+        client = FakeTaskClient(editor_mode="gibberish")
+        result, _, _ = self._run(client)
+        self.assertTrue(result.ok)  # 回落的 base 产物仍可交付
+        self.assertEqual(set(result.mode_result.fell_back), {"inv1", "dom1", "agg1"})
+        self.assertIsNone(result.library_entry)
+
+    def test_selector_gibberish_falls_back(self):
+        client = FakeTaskClient(selector_mode="gibberish")
+        result, _, _ = self._run(client)
+        self.assertTrue(result.ok)
+        self.assertTrue(result.selection.fallback)
+
+
+class TestEditorFormatRetry(unittest.TestCase):
+    """编辑格式纠正：notes-only 先重试再落盘；显式无改动一次通过；
+    两次都 notes-only 则如实抛错（阶段回落，不静默采纳）。"""
+
+    def _editor(self, mode):
+        tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(tmp.cleanup)
+        ctx = Path(tmp.name)
+        (ctx / "harness" / "scripts").mkdir(parents=True)
+        for rel, content in BASE_TREE.items():
+            (ctx / "harness" / rel).write_text(content)
+        client = FakeTaskClient(editor_mode=mode)
+        editor = CodexEditor(client, project_id="p", run_tag="fmt")
+        return editor, ctx, client
+
+    def test_notes_only_then_fix_lands(self):
+        editor, ctx, client = self._editor("notes-only-then-fix")
+        result = editor.edit("You are editing an agent harness", ctx, "inv1.a1")
+        self.assertNotIn("if a > 0", result.tree["scripts/summarize.py"])
+        self.assertEqual(client.created, 2)  # 首发 + 一次格式纠正
+
+    def test_no_change_marker_keeps_base(self):
+        editor, ctx, client = self._editor("no-change")
+        result = editor.edit("You are editing an agent harness", ctx, "inv1.a1")
+        self.assertEqual(result.tree["scripts/summarize.py"], BASE_SCRIPT)
+        self.assertEqual(client.created, 1)
+
+    def test_notes_only_twice_raises(self):
+        editor, ctx, client = self._editor("notes-only")
+        with self.assertRaises(ValueError):
+            editor.edit("You are editing an agent harness", ctx, "inv1.a1")
+        self.assertEqual(client.created, 2)
+
+    def test_parse_no_change(self):
+        self.assertEqual(parse_no_change("<no-change>no defect</no-change>"),
+                         "no defect")
+        self.assertEqual(parse_no_change("just notes"), "")
+
+
+class TestUnwrapResultText(unittest.TestCase):
+    """拆信封：信封取 text；其它原样返回，不丢内容。"""
+
+    def test_envelope_unwrapped(self):
+        raw = json.dumps({"task": "t1", "status": "completed",
+                          "text": "=== HARNESS FILE: a.md ===\nhi\n=== END HARNESS FILE ===\n"})
+        self.assertTrue(unwrap_result_text(raw).startswith("=== HARNESS FILE"))
+
+    def test_plain_text_unchanged(self):
+        self.assertEqual(unwrap_result_text("Total: 85.15"), "Total: 85.15")
+        self.assertEqual(unwrap_result_text('{"no": "text field"}'),
+                         '{"no": "text field"}')
+        self.assertEqual(unwrap_result_text("{not json"), "{not json")
+        self.assertEqual(unwrap_result_text(""), "")
+
+
+class TestEnvelopeFullRound(unittest.TestCase):
+    """信封形状整轮：假控制面按真实信封返回时，编辑块仍能落盘、
+    探针交付物是干净文本（而非信封 JSON），经验正常入库。"""
+
+    def test_envelope_round(self):
+        client = FakeTaskClient(envelope=True)
+        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="env",
+            work_dir=root / "work", domain="accounting",
+            source_files={"ledger.csv": LEDGER},
+            k=1, mode="parallel", library_root=root / "library")
+        self.assertTrue(result.ok)
+        self.assertEqual(result.mode_result.rollouts["agg1"].deliverable_text,
+                         "Total: 85.15")
+        self.assertNotIn("thread_id",
+                         result.mode_result.rollouts["agg1"].deliverable_text)
+        self.assertIsNotNone(result.library_entry)
+
+
+class TestHttpClient(unittest.TestCase):
+    def test_client_against_local_server(self):
+        seen = {"keys": [], "uploads": [], "tasks": []}
+        polls = {"t1": 0}
+
+        class Handler(BaseHTTPRequestHandler):
+            def log_message(self, *args):
+                pass
+
+            def _json(self, status, obj):
+                raw = json.dumps(obj).encode()
+                self.send_response(status)
+                self.send_header("Content-Type", "application/json")
+                self.send_header("Content-Length", str(len(raw)))
+                self.end_headers()
+                self.wfile.write(raw)
+
+            def do_POST(self):
+                seen["keys"].append(self.headers.get("X-API-Key"))
+                length = int(self.headers.get("Content-Length") or 0)
+                body = self.rfile.read(length) if length else b""
+                if self.path == "/uploads":
+                    seen["uploads"].append(body)
+                    self._json(201, {"upload_id": "u1", "files": []})
+                elif self.path == "/tasks":
+                    seen["tasks"].append(json.loads(body.decode()))
+                    self._json(201, {"id": "t1"})
+                elif self.path == "/tasks/t1/cancel":
+                    self._json(200, {})
+                else:
+                    self._json(404, {"error": "not found"})
+
+            def do_GET(self):
+                if self.path == "/tasks/t1":
+                    polls["t1"] += 1
+                    status = "queued" if polls["t1"] == 1 else "completed"
+                    self._json(200, {"id": "t1", "status": status,
+                                     "result_text": "Total: 85.15" if status == "completed" else "",
+                                     "error_text": ""})
+                elif self.path == "/tasks/t1/cost":
+                    self._json(200, {"task_id": "t1", "usd_micros": 2500000})
+                elif self.path == "/tasks/t1/trace":
+                    self._json(200, {"entries": [{"id": "e1", "kind": "agent",
+                                                  "text": "hello"}]})
+                else:
+                    self._json(404, {"error": "not found"})
+
+        server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+        thread = threading.Thread(target=server.serve_forever, daemon=True)
+        thread.start()
+        try:
+            client = HttpTaskClient(f"http://127.0.0.1:{server.server_address[1]}",
+                                    "test-key")
+            done = run_task(client, project_id="p1", prompt="do it",
+                            files={"a.txt": b"hi"}, poll_seconds=0.01,
+                            timeout_seconds=10)
+        finally:
+            server.shutdown()
+            server.server_close()
+        self.assertEqual(done.state.status, "completed")
+        self.assertEqual(done.state.result_text, "Total: 85.15")
+        self.assertEqual(done.cost_usd, 2.5)
+        self.assertIn("hello", done.trace_text)
+        self.assertTrue(all(k == "test-key" for k in seen["keys"]))
+        self.assertEqual(seen["tasks"][0]["project_id"], "p1")
+        fields, files = parse_multipart(seen["uploads"][0],
+                                        "multipart/form-data; boundary=" +
+                                        seen["uploads"][0].split(b"\r\n")[0][2:].decode())
+        self.assertEqual(dict(files), {"a.txt": b"hi"})
+
+
+class TestHttpClientRetry(unittest.TestCase):
+    """瞬断重试：掉连接与 503 各重试一次后成功；401 只请求一次。"""
+
+    def _serve(self, handler_cls):
+        server = ThreadingHTTPServer(("127.0.0.1", 0), handler_cls)
+        thread = threading.Thread(target=server.serve_forever, daemon=True)
+        thread.start()
+        self.addCleanup(server.shutdown)
+        self.addCleanup(server.server_close)
+        return HttpTaskClient(
+            f"http://127.0.0.1:{server.server_address[1]}", "test-key",
+            timeout=5)
+
+    def test_drop_then_success(self):
+        counts = {"n": 0}
+
+        class Handler(BaseHTTPRequestHandler):
+            def log_message(self, *args):
+                pass
+
+            def do_GET(self):
+                counts["n"] += 1
+                if counts["n"] == 1:
+                    self.close_connection = True
+                    return
+                raw = json.dumps({"ok": True}).encode()
+                self.send_response(200)
+                self.send_header("Content-Type", "application/json")
+                self.send_header("Content-Length", str(len(raw)))
+                self.end_headers()
+                self.wfile.write(raw)
+
+        client = self._serve(Handler)
+        self.assertEqual(client._request("GET", "/flaky"), {"ok": True})
+        self.assertEqual(counts["n"], 2)
+
+    def test_503_then_success(self):
+        counts = {"n": 0}
+
+        class Handler(BaseHTTPRequestHandler):
+            def log_message(self, *args):
+                pass
+
+            def do_GET(self):
+                counts["n"] += 1
+                raw = json.dumps({"ok": counts["n"] > 1}).encode()
+                self.send_response(503 if counts["n"] == 1 else 200)
+                self.send_header("Content-Type", "application/json")
+                self.send_header("Content-Length", str(len(raw)))
+                self.end_headers()
+                self.wfile.write(raw)
+
+        client = self._serve(Handler)
+        self.assertEqual(client._request("GET", "/flaky"), {"ok": True})
+        self.assertEqual(counts["n"], 2)
+
+    def test_401_not_retried(self):
+        counts = {"n": 0}
+
+        class Handler(BaseHTTPRequestHandler):
+            def log_message(self, *args):
+                pass
+
+            def do_GET(self):
+                counts["n"] += 1
+                raw = json.dumps({"error": "unauthorized"}).encode()
+                self.send_response(401)
+                self.send_header("Content-Type", "application/json")
+                self.send_header("Content-Length", str(len(raw)))
+                self.end_headers()
+                self.wfile.write(raw)
+
+        client = self._serve(Handler)
+        with self.assertRaises(RuntimeError):
+            client._request("GET", "/flaky")
+        self.assertEqual(counts["n"], 1)
+
+
+if __name__ == "__main__":
+    unittest.main()

docs/harness2.md

@@ -0,0 +1,184 @@
+# Harness² 融合说明
+
+状态：方法层（`cloud/harness2.py` + `cloud/tests/test_harness2.py` +
+`examples/harness2_demo.py`）与接线层（`cloud/harness2_tasks.py` +
+`cloud/tests/test_harness2_tasks.py` + `examples/harness2_run.py`）
+均已落地，在独立分支 `feat-harness2`（已 rebase 到生产 #133 的
+c757810）上，不碰主线与生产，**未合并、未部署**。控制台评审页
+与 `auto_apply` 开关仍是下一步，见文末。
+
+## 上游是什么
+
+Google Research 的 harness2（Apache 2.0，arXiv:2610.14224）：模型权重
+不动，只优化模型周围的 harness——系统提示、guardrails、记忆、工具
+指引、子代理、技能、脚本、插件。两层递归：
+
+- 任务层：Propose–Probe–Compose。对同一任务提 INV（通用做法）与 DOM
+  （领域做法）两类候选 harness，每个候选真实跑一次（探针），合成器
+  只把有行为证据支撑的改动合成终版；没有标准答案，靠行为对比。
+- 经验层：采纳后反思出一条有证据的经验（cue / lesson / evidence），
+  连同被采纳 harness 的完整树存进分领域经验库，供后续任务参考。
+
+上游报告：单步改进在三个专业 benchmark 上最高 +11.5 / +4.5 /
++11.8 个点；k=5 并行递归在 WorkBuddy 上到 80.8。这些数字是上游
+benchmark 上的，不是本项目的承诺——本项目没有 benchmark，接线后
+必须先在自己的重复任务集上跑 base 对照，才能谈收益。
+
+## 融合映射：照搬 / 改写 / 不引
+
+| 上游东西 | 本项目做法 |
+|---|---|
+| c1–c8 可编辑树（codex surface） | 照搬：AGENTS.md（8000 字符上限）、`.agents/skills/*/SKILL.md`、`.agents/skills/*/scripts/*`、`sub_agents/*/AGENT.md`、`scripts/*.py`、`.codex/hooks.json`；白名单外一律不读不写；校验含技能 description、触发行、脚本 compile、脚本引用存在、钩子 JSON 与事件名 |
+| Propose–Probe–Compose，parallel / sequential | 照搬语义，改写实现：`run_parallel` / `run_sequential`，提议被拒给一次按反馈修正的机会，再拒回落到当前 harness；探针失败记为死探针，不炸整轮 |
+| 匿名选择（只比产物） | 照搬：终版产物按任务+k+模式定种子洗牌匿名化，选择器看不到 harness 来源；选失败回落到已交付终版；k=1 无选择器时 trivial 直出 |
+| 经验库（分领域、整树 + EXPERIENCE.md + INDEX.md） | 照搬存储格式，改写生效语义：入库 ≠ 自动生效（见下） |
+| Ledger（会话/探针账本）与成本 | 改写：成本汇总进 `ModeResult.total_cost_usd`，由调用方接现有 `cloud/cost.py` / 租户配额计费；另有 `budget_usd` 超支门：超支后不再调用编辑器/探针，剩余阶段回落，且该轮不入库 |
+| 提示词 | 改写短版（中文项目语境下的英文任务提示，结构同上游，措辞自写），不抄上游长提示 |
+| Benchmark 适配器（JobBench / LAB / WorkBuddy）、评分器、rubric | 不引。任务就是一个普通 codex 任务，delivered 由 adapter 定义（如产物存在且非空） |
+| Proposer 通道（Claude CLI / OpenCode）、Vertex 配置 | 不引。编辑器/选择器/反思器是调用方注入的协议，真实接线时是 codex 任务（见下） |
+| 上游执行默认：网络不受限、Codex sandbox 禁用 | 明确不引，见安全一节 |
+| 上游 ThreadPool 并行探针 | 不引。核心循环串行跑探针；并行是调用方在 adapter 层把每个探针发成独立 codex 任务来实现，避免核心循环抢同一个 Worker/DB |
+
+## 安全偏差（刻意与上游不同，勿“纠回”上游）
+
+1. **执行面与本模块解耦。** `cloud/harness2.py` 不跑任何命令、不拼
+   任何 codex exec 参数。真实 adapter 必须走 Worker 已有的 Docker /
+   gVisor 沙箱、租户隔离与 egress 白名单，绝不加
+   `--dangerously-bypass-approvals-and-sandbox`。上游 README 明写其
+   Codex 执行是 sandbox 禁用、网络不受限——那是 benchmark 实验的
+   默认，不是生产服务的默认。
+2. **采纳 ≠ 自动铺开。** 多租户下，一个编辑器写出的 AGENTS.md 若
+   自动改写别人的任务，是质量与越权双重风险。所以 `improve_task`
+   的 `auto_apply` 默认关：库只是“待评审的经验”，提议器可读，
+   但任何任务的 base 永远是项目自己的 base harness；只有项目显式
+   开启 auto_apply 后，调用方才在下一轮的 `base_harness()` 里取库
+   中最新采纳树。这个开关的服务端接线（项目设置 + 控制台评审页）
+   是下一步。
+3. **预算是硬门。** k 硬上限 3；parallel 一轮最多 10 次执行
+   （1 base + 2k 候选 + k 终版），sequential 同样最多 10 次。
+   `budget_usd` 超支后不再花钱，且该轮不入库——超支轮次的“经验”
+   是不完整证据，不配进库。
+4. **编辑器写出的东西先校验后采纳。** 校验失败的提议给一次修正
+   机会，再失败就回落，绝不把坏树（空 AGENTS.md、坏钩子、不编译
+   的脚本、无触发行的技能）铺到探针里。
+
+## 接口（调用方视角）
+
+```python
+from cloud.harness2 import improve_task, Library
+from pathlib import Path
+
+result = improve_task(adapter, editor,
+                      selector=selector, reflector=reflector,
+                      work_dir=Path("runs/h2-task-1"), k=1, mode="parallel",
+                      library=Library(Path("runs/library"), domain="accounting"),
+                      budget_usd=5.0)
+result.to_dict()  # 阶段、回落、选择、经验、入库位置、总成本
+```
+
+- `TaskAdapter`：`task_id / domain / prompt` 属性；`base_harness()`；
+  `execute(label, harness) -> Rollout`（失败也返回 Rollout，不抛）；
+  可选 `validate_tree()` 做项目特有校验。
+- `Editor.edit(mission, context_dir, label) -> EditResult`：读
+  `context_dir` 下的 `task.md / rollouts / deliverables / harness /
+  library`，改 `context/harness/`，返回改完的树（`tree=None` 时
+  框架自动从 `context/harness/` 读回——真实 codex 任务接线就用这条
+  路）；`notes` 是编辑器最终消息里 `<notes>` 块的解析结果，由调用
+  方在构造 `EditResult` 时填好。
+- `Selector.select(mission, context_dir) -> (id, why, cost)`：读
+  `context_dir/candidates/<id>.md`（匿名产物），返回 choice id。
+- `Reflector.reflect(mission, context_dir) -> dict`：返回含 cue /
+  lesson / evidence（可含 confidence / edits）的经验，三项不全
+  返回 `{}` 即表示无可入库经验。
+
+证据上下文由 `build_context` 生成：证据以文件形式给模型，prompt
+只写指针，不把轨迹截断后塞进 prompt（截断过的轨迹会编造缺陷，
+这是上游明确反对的做法）。
+
+## 接线层（已落地，`cloud/harness2_tasks.py`）
+
+只用控制面已有的 API，不新增任何服务端端点：`POST /uploads` →
+`POST /tasks`（`upload_id` 绑定 + `idempotency_key` 幂等）→
+`GET /tasks/{id}` 轮询终态 → `GET /tasks/{id}/trace` 读轨迹 →
+`GET /tasks/{id}/cost` 读 `usd_micros` → 超时 `POST /tasks/{id}/cancel`。
+认证是 `X-API-Key` 请求头；密钥只由调用方持有，接线层不读、不存、
+不打印密钥。
+
+- **探针**：bundle = 源文件 + harness 树，Worker 派发时经
+  `host.py --seed-dir` 拷进 worktree，harness 文件覆盖仓库同名文件
+  ——同一任务在不同 harness 下跑。prompt = 原任务 prompt + 交付脚注。
+- **结果读取**：控制面返回的 `result_text` 是一层 JSON 信封
+  （task / thread_id / cwd / status / text / usage），agent 的最终
+  回复在 `text` 字段里。`run_task` 是四路共用的 choke point，
+  在此用 `unwrap_result_text` 拆信封；直接拿信封当回复会让
+  行首锚定的文件块解析永远看不见块（2026-10-29 三轮真实轮次
+  的真根因：编辑明明发了块，却被判成无改动/格式错误）。
+- **编辑**：bundle = `context/` 前缀的证据树；agent 在 worktree 的
+  `context/harness/` 里改文件，最终回复用文件块（`=== HARNESS FILE:
+  路径 ===` … `=== END HARNESS FILE ===`）只输出改动文件，没改的
+  省略即继承，外加 `<notes>` 块；接线层解析后落到 fallback 树上再
+  走核心校验。格式契约（2026-10-29 两轮真实轮次后收紧）：只在
+  `<notes>` 里声称改动、却不发文件块，不再算“无需改动”——无改动
+  必须用显式 `<no-change>理由</no-change>` 标记；文件块与标记同
+  时出现、路径越界、块未闭合都算格式错误，接线层用独立幂等键
+  带反馈重试一次（`.fmt2`），两次都错才如实回落，绝不静默采纳。
+- **选择 / 反思**：同款 bundle + 最终回复解析（` ```choice ` +
+  `<why>`；`<experience>` + `<edits>`），不读 worktree 文件。
+- **幂等**：`idempotency_key = h2-{run_tag}-{role}-{label}`，同一
+  run_tag 重跑复用已有任务（断点续跑）；想从头重跑必须换 run_tag，
+  换了 harness 内容却沿用旧 run_tag 时旧结果会被复用、白跑一轮。
+- **成本与审批**：接线层不拼任何 codex 参数、不改任务默认值；
+  model / effort / approval / environment_id / executor 全由调用方
+  透传，缺省走 Worker 默认值；任务走与普通任务完全相同的执行器、
+  沙箱、租户隔离、配额门与审计；单任务超时默认 3600 秒，超时先
+  取消再记失败/回落；长任务走 #133 的自动续跑。
+- **接线层最大的适用边界**：探针交付物 = 最终回复文本。文件型
+  产物（代码 diff、报告文件）目前只能靠 agent 在最终回复里复述/
+  摘要：worktree 快照 API 只给 diff 截断（20000 字）与未跟踪文件
+  列表、不给完整文件内容，跨主机还读不到。文件产物评审需要新增
+  文件读取端点，另行评审，不在本层假装已解决。
+
+用真实任务跑一轮（先 dry-run 看计划，确认后加 `--yes`）：
+
+```bash
+export CODEX_API_KEY='...'   # 密钥只走环境变量，不接受命令行参数
+python3 examples/harness2_run.py --base-url http://127.0.0.1:8765 \
+    --project-id <测试项目> --prompt-file task.md \
+    --base-harness-dir ./my-base-harness --source-dir ./task-source \
+    --mode parallel --k 1 --budget 5 --output runs/h2-real-1
+# 计划无误后，加 --yes 重跑；同一 run_tag 可断点续跑
+```
+
+## 仍未做的事（需另行评审）
+
+1. **人工关口**：采纳的 harness 先在控制台以“待评审经验”展示，
+   项目负责人点“采纳为 base”后才影响后续任务；`auto_apply` 开关
+   默认关，开启需项目负责人显式操作并记审计。目前
+   `run_improvement_via_tasks` 的 `auto_apply` 恒为 False，库只是
+   待评审经验。
+2. **文件产物评审**：新增读取已终态任务 worktree 文件的端点（权限、
+   跨主机、大小上限都要定），或让探针任务把产物文件内容写进最终
+   回复的约定成为正式契约——前者是服务端改动，后者是提示词契约，
+   各有利弊，需评审。
+3. **效果验证**：先在 3–5 个重复性任务（如固定格式的对账、周报
+   生成）上跑 base 对照：同一任务、同一模型，base 产物 vs 改进后
+   产物由人工或固定检查脚本判定，记录执行次数与成本；没有对照
+   就不宣称收益。
+
+## 演示与测试（无模型、无网络）
+
+```bash
+python3 examples/harness2_demo.py --mode parallel    # 应输出 Total: 85.15 并入库
+python3 examples/harness2_demo.py --mode sequential
+python3 -m unittest discover -s cloud/tests -p "test_harness2*.py" -v  # 25 项
+```
+
+接线层单测用进程内假客户端（探针真跑 bundle 里的脚本、编辑按
+文件块格式回改动）跑通完整轮次，并对本地假服务端实测了 HTTP
+客户端（建任务、轮询、成本、轨迹、密钥头、multipart 与服务端
+解析器往返）。真实任务链路尚未实测：第一次真实运行请用测试项目，
+并先跑 `examples/harness2_run.py` 不带 `--yes` 的计划模式。
+
+演示账本 100.25 / -20.10 / 5.00：base 脚本只加正数得 105.25，
+改进后带符号和为 85.15；演示用的编辑器/选择器/反思器是确定性的，
+只为证明循环本身，真实收益必须用第 4 步的对照来量。

examples/harness2_demo.py

@@ -0,0 +1,138 @@
+"""Harness² 融合演示：不花模型钱，跑通一轮完整改进。
+
+用法（在项目根目录）：
+    python3 examples/harness2_demo.py --mode parallel --k 1
+    python3 examples/harness2_demo.py --mode sequential --k 1 --output /tmp/h2demo
+
+任务是对账本求和。base harness 的脚本只加正数（账本含一笔负数 credit，
+base 会漏掉它）；确定性编辑器去掉正数过滤；选择器按产物挑带符号和；
+反思器把“正负都要算”写进经验库。跑完打印：被采纳的产物、回落情况、
+经验库条目位置。真实接线时，这里的三个确定性类换成 codex 任务
+（见 docs/harness2.md），执行器换成 Worker 的任务执行，循环本身不变。
+"""
+
+import argparse
+import csv
+import subprocess
+import sys
+import tempfile
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from cloud.harness2 import (  # noqa: E402
+    EditResult, Library, Rollout, improve_task, read_tree, write_tree,
+)
+
+BASE_SCRIPT = """import csv
+import sys
+from decimal import Decimal
+with open(sys.argv[1]) as f:
+    amounts = [Decimal(r['amount']) for r in csv.DictReader(f)]
+print(f"Total: {sum((a for a in amounts if a > 0), Decimal(0)):.2f}")
+"""
+
+BASE_TREE = {
+    "AGENTS.md": "Use scripts/summarize.py to total the ledger.\n",
+    "scripts/summarize.py": BASE_SCRIPT,
+}
+
+
+class LedgerAdapter:
+    task_id = "ledger-1"
+    domain = "accounting"
+    prompt = "Sum every signed amount in ledger.csv. Return Total: with two decimals."
+
+    def __init__(self, root):
+        self.root = Path(root)
+        self.root.mkdir(parents=True, exist_ok=True)
+        self.ledger = self.root / "ledger.csv"
+        self.ledger.write_text("amount\n100.25\n-20.10\n5.00\n", encoding="utf-8")
+        self.executions = 0
+
+    def base_harness(self):
+        return dict(BASE_TREE)
+
+    def execute(self, label, harness):
+        self.executions += 1
+        work = self.root / "runs" / label
+        work.mkdir(parents=True, exist_ok=True)
+        write_tree(harness, work / "harness")
+        proc = subprocess.run(
+            [sys.executable, str(work / "harness" / "scripts" / "summarize.py"),
+             str(self.ledger)], capture_output=True, text=True, timeout=10)
+        out = proc.stdout.strip()
+        return Rollout(label=label, ok=proc.returncode == 0,
+                       delivered=out.startswith("Total:"), deliverable_text=out,
+                       trajectory="[ACTION] Run scripts/summarize.py\n[OBSERVATION]\n" + out,
+                       cost_usd=0.01)
+
+    def validate_tree(self, harness):
+        if "scripts/summarize.py" not in harness:
+            raise ValueError("The local executor requires scripts/summarize.py")
+
+
+class FixEditor:
+    def edit(self, mission, context_dir, label):
+        harness_dir = Path(context_dir) / "harness"
+        script = harness_dir / "scripts" / "summarize.py"
+        script.write_text(script.read_text().replace(" if a > 0", ""), encoding="utf-8")
+        return EditResult(tree=read_tree(harness_dir),
+                          notes="Keep signed amounts: the base total omitted credits.")
+
+
+class PickSignedSelector:
+    def select(self, mission, context_dir):
+        for path in sorted((Path(context_dir) / "candidates").glob("*.md")):
+            if "Total: 85.15" in path.read_text():
+                return path.stem, "The signed total matches the source ledger.", 0.0
+        raise ValueError("no candidate with the signed total")
+
+
+class LedgerReflector:
+    def reflect(self, mission, context_dir):
+        root = Path(context_dir)
+        changed = (root / "harness" / "scripts" / "summarize.py").read_text() != (
+            root / "base_harness" / "scripts" / "summarize.py").read_text()
+        if not changed:
+            return {}
+        return {
+            "cue": "When a ledger includes credits and debits",
+            "lesson": "Include both signs when summing in scripts/summarize.py.",
+            "evidence": "Removing the positive-only filter changed the probe total to "
+                        "include the credits omitted by the base trajectory.",
+            "confidence": "high for this observed arithmetic contrast",
+        }
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--mode", default="parallel", choices=["parallel", "sequential"])
+    ap.add_argument("--k", type=int, default=1)
+    ap.add_argument("--output", default="")
+    args = ap.parse_args()
+    tmp = None
+    root = Path(args.output) if args.output else Path(tempfile.mkdtemp(prefix="h2demo-"))
+    if not args.output:
+        tmp = root
+    adapter = LedgerAdapter(root)
+    library = Library(root / "library", "accounting")
+    result = improve_task(adapter, FixEditor(), selector=PickSignedSelector(),
+                          reflector=LedgerReflector(), work_dir=root / "work",
+                          k=args.k, mode=args.mode, library=library)
+    chosen = result.mode_result.rollouts[result.selection.chosen_label]
+    print(f"mode={result.mode} k={args.k} ok={result.ok}")
+    print(f"executions={adapter.executions} (base + candidates + finals)")
+    print(f"chosen={result.selection.chosen_label} deliverable={chosen.deliverable_text}")
+    print(f"fell_back={result.mode_result.fell_back or 'none'}")
+    print(f"library_entry={result.library_entry}")
+    print(f"experience={result.experience.get('lesson', '(none)')}")
+    print(f"work_dir={root}")
+    assert chosen.deliverable_text == "Total: 85.15", "demo must end at the signed total"
+    assert result.library_entry is not None, "demo must enter the library"
+    print("demo OK: signed total adopted and stored in the library")
+    _ = tmp
+
+
+if __name__ == "__main__":
+    main()

examples/harness2_run.py

@@ -0,0 +1,133 @@
+"""用真实 codex 任务跑一轮 Harness² 改进（接线层 CLI）。
+
+这会创建真实任务、花真实模型配额——所以默认只打印计划，不建任务：
+先跑一遍不带 --yes 看计划，确认项目、任务数与预算无误后，再带
+--yes 重跑。第一次请务必用测试项目，不要直接指生产项目。
+
+密钥不走命令行参数（ps 可见）：用环境变量 CODEX_API_KEY。
+
+示例：
+    export CODEX_API_KEY='...'
+    python3 examples/harness2_run.py --base-url http://127.0.0.1:8765 \\
+        --project-id <test-project> --prompt-file task.md \\
+        --base-harness-dir ./my-base-harness --source-dir ./task-source \\
+        --mode parallel --k 1 --budget 5 --output runs/h2-real-1
+    # 看计划无误后，加 --yes 重跑（同一 run_tag 可断点续跑；想从头
+    # 重跑就换 --run-tag 或不传，让它自动生成新的时间戳 tag）。
+
+base-harness-dir：一个目录，里面就是 harness 树（AGENTS.md、
+.agents/、scripts/、.codex/……），按原样读入；不存在或 AGENTS.md
+为空时直接报错，不跑。
+source-dir（可选）：原任务需要的源文件，会随 harness 一起 seed 进
+每个探针任务的 worktree；与 harness 路径重名时 harness 优先。
+"""
+
+import argparse
+import os
+import sys
+import tempfile
+import time
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from cloud.harness2 import read_tree  # noqa: E402
+from cloud.harness2_tasks import (  # noqa: E402
+    HttpTaskClient, run_improvement_via_tasks,
+)
+
+
+def _flatten(root: Path) -> dict[str, bytes]:
+    out: dict[str, bytes] = {}
+    if not root or not root.is_dir():
+        return out
+    for p in sorted(root.rglob("*")):
+        if p.is_file() and not p.is_symlink() and "__pycache__" not in p.parts:
+            out[str(p.relative_to(root))] = p.read_bytes()
+    return out
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--base-url", required=True, help="控制面地址，如 http://127.0.0.1:8765")
+    ap.add_argument("--project-id", required=True)
+    ap.add_argument("--prompt", default="")
+    ap.add_argument("--prompt-file", default="")
+    ap.add_argument("--task-id", default="h2-task-1")
+    ap.add_argument("--domain", default="general")
+    ap.add_argument("--base-harness-dir", required=True)
+    ap.add_argument("--source-dir", default="")
+    ap.add_argument("--mode", default="parallel", choices=["parallel", "sequential"])
+    ap.add_argument("--k", type=int, default=1)
+    ap.add_argument("--budget", type=float, default=None)
+    ap.add_argument("--run-tag", default="")
+    ap.add_argument("--output", default="")
+    ap.add_argument("--library-root", default="")
+    ap.add_argument("--model", default=None)
+    ap.add_argument("--effort", default=None)
+    ap.add_argument("--approval", default=None)
+    ap.add_argument("--environment-id", default=None)
+    ap.add_argument("--executor", default=None)
+    ap.add_argument("--no-time-limit", action="store_true")
+    ap.add_argument("--timeout", type=int, default=3600)
+    ap.add_argument("--yes", action="store_true", help="确认创建真实任务并消费配额")
+    args = ap.parse_args()
+
+    api_key = os.environ.get("CODEX_API_KEY")
+    if not api_key:
+        print("缺少 CODEX_API_KEY 环境变量（密钥不接受命令行参数）", file=sys.stderr)
+        return 2
+    prompt = args.prompt
+    if args.prompt_file:
+        prompt = Path(args.prompt_file).read_text(encoding="utf-8")
+    if not prompt.strip():
+        print("需要 --prompt 或 --prompt-file", file=sys.stderr)
+        return 2
+    base = read_tree(args.base_harness_dir)
+    if not base.get("AGENTS.md", "").strip():
+        print(f"base-harness-dir 里没有可用的 AGENTS.md：{args.base_harness_dir}",
+              file=sys.stderr)
+        return 2
+    run_tag = args.run_tag or time.strftime("h2-%Y%m%d-%H%M%S")
+    # 任务数（k=1 parallel）：1 base + 2 候选 + 1 终版探针 + 3 编辑
+    # + 1 选择 + 1 反思 = 9；sequential 同为 9；k 越大按 3k+? 线性涨，
+    # 上限 k=3 时 parallel 为 1 + 6 + 3 探针 + 9 编辑 + 1 选择 + 1 反思 = 21。
+    n_probes = 1 + 3 * args.k  # 两种模式同款：1 base + 每轮 2 候选 + 1 终版
+    n_edits = 3 * args.k
+    plan = (f"计划：项目 {args.project_id}，任务 {args.task_id}，"
+            f"mode={args.mode} k={args.k}，run_tag={run_tag}，"
+            f"约 {n_probes} 个探针 + {n_edits} 个编辑 + 1 选择 + 1 反思 = "
+            f"{n_probes + n_edits + 2} 个真实任务，"
+            f"budget_usd={args.budget}，base 文件 {len(base)} 个。")
+    print(plan)
+    if not args.yes:
+        print("未带 --yes：只打印计划，不创建任何任务。确认无误后加 --yes 重跑。")
+        return 0
+    work_dir = Path(args.output) if args.output else Path(tempfile.mkdtemp(prefix="h2run-"))
+    client = HttpTaskClient(args.base_url, api_key)
+    result = run_improvement_via_tasks(
+        client, project_id=args.project_id, task_id=args.task_id, prompt=prompt,
+        base_harness=base, run_tag=run_tag, work_dir=work_dir, domain=args.domain,
+        source_files=_flatten(Path(args.source_dir)) if args.source_dir else None,
+        k=args.k, mode=args.mode,
+        library_root=Path(args.library_root) if args.library_root else None,
+        budget_usd=args.budget, model=args.model, effort=args.effort,
+        approval=args.approval, environment_id=args.environment_id,
+        executor=args.executor, no_time_limit=args.no_time_limit,
+        timeout_seconds=args.timeout)
+    d = result.to_dict()
+    chosen = result.mode_result.rollouts.get(result.selection.chosen_label)
+    print(f"ok={result.ok} chosen={result.selection.chosen_label} "
+          f"fallback={result.selection.fallback} "
+          f"fell_back={result.mode_result.fell_back or 'none'} "
+          f"budget_exceeded={result.budget_exceeded}")
+    print(f"deliverable={(chosen.deliverable_text if chosen else '')[:500]}")
+    print(f"total_cost_usd={result.mode_result.total_cost_usd} "
+          f"library_entry={result.library_entry}")
+    print(f"work_dir={work_dir}")
+    print(f"full_result={d}")
+    return 0 if result.ok else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())