harness2: per-role models, sweep batch, task file assets endpoint

- Role models: run_improvement_via_tasks gains role_models
  (probe/editor/selector/reflector) plus --model-<role> CLI flags;
  missing/empty falls back to shared --model, both empty means
  server default, so default behavior is unchanged.
- Sweep: run_sweep + examples/harness2_sweep.py manifest CLI with
  shared library_root (library restore), per-run failure isolation
  and sweep-report.json summary; preflight gates task creation.
- Deliverable assets: GET /tasks/{id}/files for local-executor
  tasks (safe path resolution, UTF-8 text <=256KB, listing <=200;
  docker/cloud/cross-host 409 with explicit reason, since tenant
  Docker shares one /app/work per project and per-task attribution
  would misattribute); probe rollout deliverable_files from the
  endpoint plus DELIVERABLE FILE reply blocks (endpoint wins on
  collision, bad blocks dropped), surfaced in
  deliverables/<label>/files/ context and full_result summary.
- Docs: harness2.md comparison table updated for all three items.

cloud/tests/test_harness2_tasks.py

@@ -19,26 +19,26 @@ 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,
+        parse_choice, parse_deliverable_files, parse_experience,
+        parse_harness_files, parse_no_change, parse_notes, parse_why,
     )
     from cloud.harness2_tasks import (
         CodexEditor, CodexProbeAdapter, HttpTaskClient, TaskState,
-        build_multipart, preflight_checks, run_improvement_via_tasks, run_task,
-        unwrap_result_text, validate_run_tag,
+        build_multipart, preflight_checks, run_improvement_via_tasks, run_sweep,
+        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
     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,
+        parse_choice, parse_deliverable_files, parse_experience,
+        parse_harness_files, parse_no_change, parse_notes, parse_why,
     )
     from cloud.harness2_tasks import (
         CodexEditor, CodexProbeAdapter, HttpTaskClient, TaskState,
-        build_multipart, preflight_checks, run_improvement_via_tasks, run_task,
-        unwrap_result_text, validate_run_tag,
+        build_multipart, preflight_checks, run_improvement_via_tasks, run_sweep,
+        run_task, unwrap_result_text, validate_run_tag,
     )
     from cloud.uploads import parse_multipart, parse_paths_field
 
@@ -61,15 +61,22 @@ 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):
+    def __init__(self, editor_mode="fix", selector_mode="signed", envelope=False,
+                 fail_tags=(), endpoint_files=None, endpoint_error=False,
+                 probe_extra=""):
         self.editor_mode = editor_mode
         self.selector_mode = selector_mode
         self.envelope = envelope
+        self.fail_tags = tuple(fail_tags)
+        self.endpoint_files = dict(endpoint_files or {})
+        self.endpoint_error = endpoint_error
+        self.probe_extra = probe_extra
         self.uploads = {}
         self.tasks = {}
         self.by_key = {}
         self.created = 0
         self.cancelled = []
+        self.task_models = {}
 
     def upload(self, files):
         uid = f"u{len(self.uploads) + 1}"
@@ -78,6 +85,8 @@ class FakeTaskClient:
 
     def create_task(self, *, project_id, prompt, upload_id=None,
                     idempotency_key=None, **kwargs):
+        if idempotency_key and any(t in idempotency_key for t in self.fail_tags):
+            raise RuntimeError("injected task failure")
         if idempotency_key and idempotency_key in self.by_key:
             return self.by_key[idempotency_key]
         tid = f"t{len(self.tasks) + 1}"
@@ -94,6 +103,7 @@ class FakeTaskClient:
             })
         self.tasks[tid] = TaskState(task_id=tid, status="completed",
                                     result_text=result)
+        self.task_models[tid] = kwargs.get("model")
         if idempotency_key:
             self.by_key[idempotency_key] = tid
         return tid
@@ -163,11 +173,22 @@ class FakeTaskClient:
                 [sys.executable, str(root / "scripts" / "summarize.py"),
                  str(root / "ledger.csv")],
                 capture_output=True, text=True, timeout=10)
-            return proc.stdout.strip()
+            return proc.stdout.strip() + self.probe_extra
 
     def get_task(self, task_id):
         return self.tasks[task_id]
 
+    def get_task_files(self, task_id, path=""):
+        if self.endpoint_error:
+            raise RuntimeError("409 files unavailable")
+        if not path:
+            return {"entries": [
+                {"path": p, "size": len(c.encode("utf-8"))}
+                for p, c in sorted(self.endpoint_files.items())]}
+        if path in self.endpoint_files:
+            return {"content": self.endpoint_files[path]}
+        raise RuntimeError("no such file")
+
     def get_trace_text(self, task_id):
         return "[ACTION] fake probe\n[OBSERVATION]\n" + self.tasks[task_id].result_text
 
@@ -192,6 +213,22 @@ class TestParseHelpers(unittest.TestCase):
         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")
+
+    def test_parse_deliverable_files_lenient(self):
+        text = ("=== DELIVERABLE FILE: a.md ===\nhello\n"
+                "=== END DELIVERABLE FILE ===\n"
+                "=== DELIVERABLE FILE: ../evil ===\nx\n"
+                "=== END DELIVERABLE FILE ===\n"
+                "=== DELIVERABLE FILE: b.md ===\nunterminated\n")
+        self.assertEqual(parse_deliverable_files(text), {"a.md": "hello\n"})
+        many = "".join(
+            f"=== DELIVERABLE FILE: f{i}.md ===\nx\n"
+            "=== END DELIVERABLE FILE ===\n" for i in range(10))
+        self.assertEqual(len(parse_deliverable_files(many)), 8)
+        big = ("=== DELIVERABLE FILE: big.md ===\n" + "x" * 70000 + "\n"
+               "=== END DELIVERABLE FILE ===\n")
+        content = parse_deliverable_files(big)["big.md"]
+        self.assertTrue(content.endswith("...[truncated]"))
         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"))
@@ -238,6 +275,109 @@ class TestFullRunViaTasks(unittest.TestCase):
         self.assertIsNotNone(result.library_entry)
         self.assertTrue((root / "library" / "accounting" / "INDEX.md").is_file())
 
+    def _models_by_role(self, client):
+        roles = {}
+        for key, tid in client.by_key.items():
+            for role in ("probe", "edit", "select", "reflect"):
+                if f"-{role}-" in key or key.endswith(f"-{role}"):
+                    roles.setdefault(role, set()).add(client.task_models[tid])
+        return roles
+
+    def test_role_models_land_on_each_role(self):
+        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="roles1",
+            work_dir=root / "work", domain="accounting",
+            source_files={"ledger.csv": LEDGER},
+            k=2, mode="parallel", library_root=root / "library",
+            model="shared-model",
+            role_models={"probe": "probe-model", "editor": "editor-model",
+                         "selector": "selector-model",
+                         "reflector": "reflector-model"})
+        self.assertTrue(result.ok)
+        roles = self._models_by_role(client)
+        self.assertEqual(roles["probe"], {"probe-model"})
+        self.assertEqual(roles["edit"], {"editor-model"})
+        self.assertEqual(roles["select"], {"selector-model"})
+        self.assertEqual(roles["reflect"], {"reflector-model"})
+
+    def test_role_model_falls_back_to_shared_then_default(self):
+        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="roles2",
+            work_dir=root / "work", domain="accounting",
+            source_files={"ledger.csv": LEDGER},
+            k=1, mode="parallel", library_root=root / "library",
+            model="shared-model",
+            role_models={"editor": "editor-model"})
+        self.assertTrue(result.ok)
+        roles = self._models_by_role(client)
+        self.assertEqual(roles["probe"], {"shared-model"})
+        self.assertEqual(roles["edit"], {"editor-model"})
+        self.assertEqual(roles["reflect"], {"shared-model"})
+        # 两者都空时不传模型（服务端默认），假客户端记为 None。
+        client2 = FakeTaskClient()
+        result2, _, _ = self._run(client2, run_tag="roles3")
+        self.assertTrue(result2.ok)
+        self.assertEqual(set(client2.task_models.values()), {None})
+
+    def test_probe_deliverable_files_from_endpoint(self):
+        client = FakeTaskClient(endpoint_files={
+            "report.md": "# R\n",
+            "ledger.csv": "amount\n999.00\n",
+            "AGENTS.md": BASE_TREE["AGENTS.md"],
+        })
+        result, _, root = self._run(client, run_tag="assets1")
+        self.assertTrue(result.ok)
+        agg = result.mode_result.rollouts["agg1"]
+        # 新增文件与改动过的 seed 文件进产物；内容与 seed 一致的不进。
+        self.assertEqual(agg.deliverable_files,
+                         {"report.md": "# R\n",
+                          "ledger.csv": "amount\n999.00\n"})
+        found = [p for p in root.rglob("report.md")
+                 if "deliverables/agg1/files" in str(p)]
+        self.assertTrue(found)
+        self.assertEqual(found[0].read_text(encoding="utf-8"), "# R\n")
+
+    def test_probe_deliverable_files_from_reply_blocks(self):
+        block = ("\n=== DELIVERABLE FILE: page.html ===\n<h1>hi</h1>\n"
+                 "=== END DELIVERABLE FILE ===\n")
+        client = FakeTaskClient(endpoint_error=True, probe_extra=block)
+        result, _, _ = self._run(client, run_tag="assets2")
+        self.assertTrue(result.ok)
+        self.assertEqual(
+            result.mode_result.rollouts["agg1"].deliverable_files,
+            {"page.html": "<h1>hi</h1>\n"})
+
+    def test_probe_deliverable_files_endpoint_wins_on_collision(self):
+        block = ("\n=== DELIVERABLE FILE: page.html ===\nfrom-reply\n"
+                 "=== END DELIVERABLE FILE ===\n")
+        client = FakeTaskClient(
+            endpoint_files={"page.html": "from-endpoint\n"},
+            probe_extra=block)
+        result, _, _ = self._run(client, run_tag="assets3")
+        self.assertTrue(result.ok)
+        self.assertEqual(
+            result.mode_result.rollouts["agg1"].deliverable_files,
+            {"page.html": "from-endpoint\n"})
+
+    def test_probe_deliverable_files_endpoint_error_is_not_fatal(self):
+        client = FakeTaskClient(endpoint_error=True)
+        result, _, _ = self._run(client, run_tag="assets4")
+        self.assertTrue(result.ok)
+        self.assertEqual(
+            result.mode_result.rollouts["agg1"].deliverable_files, {})
+
     def test_same_run_tag_replays_without_new_tasks(self):
         client = FakeTaskClient()
         result1, _, _ = self._run(client, work_sub="work1")
@@ -563,5 +703,56 @@ class TestHttpClientRetry(unittest.TestCase):
         self.assertEqual(counts["n"], 1)
 
 
+class TestSweep(unittest.TestCase):
+    def _spec(self, tag, **over):
+        spec = {
+            "run_tag": tag, "task_id": "ledger-1",
+            "prompt": "Sum every signed amount in ledger.csv. Return Total: with two decimals.",
+            "base_harness": dict(BASE_TREE),
+            "source_files": {"ledger.csv": LEDGER},
+            "k": 1, "mode": "parallel", "domain": "accounting",
+        }
+        spec.update(over)
+        return spec
+
+    def test_sweep_isolates_failures_and_shares_library(self):
+        client = FakeTaskClient(fail_tags=("bad1",))
+        tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(tmp.cleanup)
+        root = Path(tmp.name)
+        report = run_sweep(
+            client, project_id="p1",
+            runs=[self._spec("good1"), self._spec("bad1"),
+                  self._spec("badk", k=99), self._spec("good2")],
+            work_root=root / "sweep", library_root=root / "library")
+        by_tag = {e["run_tag"]: e for e in report["runs"]}
+        self.assertTrue(by_tag["good1"]["ok"])
+        self.assertEqual(by_tag["good1"]["chosen_label"], "agg1")
+        # bad1：全部任务注入失败 → 轮次 ok=False，但不中断后续轮。
+        self.assertFalse(by_tag["bad1"]["ok"])
+        # badk：k 越界在方法层抛错 → 记入 error，同样不中断。
+        self.assertFalse(by_tag["badk"]["ok"])
+        self.assertIn("ValueError", by_tag["badk"]["error"])
+        self.assertTrue(by_tag["good2"]["ok"])
+        self.assertEqual(report["totals"]["n_runs"], 4)
+        self.assertEqual(report["totals"]["n_ok"], 2)
+        # 共享库：两轮好结果都入库，同一 INDEX 可见。
+        self.assertIsNotNone(by_tag["good1"]["library_entry"])
+        self.assertIsNotNone(by_tag["good2"]["library_entry"])
+        self.assertTrue((root / "library" / "accounting" / "INDEX.md").is_file())
+
+    def test_sweep_requires_run_tag(self):
+        client = FakeTaskClient()
+        tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(tmp.cleanup)
+        root = Path(tmp.name)
+        spec = self._spec("good1")
+        del spec["run_tag"]
+        report = run_sweep(client, project_id="p1", runs=[spec],
+                           work_root=root / "sweep")
+        self.assertFalse(report["runs"][0]["ok"])
+        self.assertIn("run_tag", report["runs"][0]["error"])
+
+
 if __name__ == "__main__":
     unittest.main()