benchmark: integrate recovery fixes for continued EdgeBench work

benchmark/README.md

@@ -719,6 +719,14 @@ Omit the controller file and `controller_handoff` on a new run to disable automa
 controller capture; removing the checkpoint request disables capture entirely.
 Existing artifacts remain inert, and the source trial is not restarted by disabling it.
 
+The native handoff accepts and reads one bounded request before validating its
+physical capture root. A missing or symlinked root closes that connection;
+the original source can correct its path and continue through the same listener.
+The refusal neither captures nor releases the old packet and does not renew the
+original wake deadline. The transport does not create or follow a foreign root.
+Host failures before a connection is accepted remain execution failures. They
+are not packet refusals and cannot enter the source-preserving retry loop.
+
 To reproduce **joint source capture and both fresh-guest receivers**, use
 [`qualify_checkpoint_joint.py`](tests/qualify_checkpoint_joint.py) from the current
 clean dedicated LoopX checkout; keep it frozen for the whole command. Run the
@@ -831,6 +839,21 @@ restore boundary, and a terminal Goal still requires explicit resume. This PR
 prevents new planning-clock injection; it does not qualify old-archive recovery
 or claim improved model behavior or scores.
 
+Only the explicit `loopx-planned` benchmark worker adds research-planning
+guidance around the unchanged product planning packet: check decision-critical
+assumptions, compare with and retain the strongest available reference, and keep
+an informative negative experiment separate from Goal completion. This is a
+provider planning treatment, not new default Core/CLI guidance or a checkpoint
+requirement. Ordinary Goal entry, seeded work and official native workers do not
+use this wrapper. Pin this provider revision when comparing runs; the guidance
+does not establish model adoption, initial-quality parity or long-horizon gain.
+
+仅显式 `loopx-planned` benchmark worker 在未修改的产品规划 packet 外附加研究
+指引：核对关键假设、对照并保留最强起点、区分有信息量的负实验与 Goal 完成。
+这是 provider 规划处理，不是 Core/CLI 默认指引或 checkpoint 要求。普通 Goal、
+预置任务和 official 原生 worker 不使用此包装；对比运行须固定 provider revision。
+这些指引不证明模型采用、首次质量持平或长程收益。
+
 The sealed deadline and native capture timestamp determine the remainder; the
 receiver recomputes it. A point with two hours remaining keeps that duration
 after an eight-hour offline gap. It does not reuse the old absolute epoch or
@@ -935,6 +958,11 @@ worker 保留原 guard，controller 暂停全部容器写入者，核对仍有
 文件 witness，再封存原生评测机会、incumbent 及剩余预算。源变化、过期或失去
 交接会拒绝；可恢复的捕获失败不取消原研究，也不得改用稍后的状态。解除暂停后，
 源会话继续使用原 packet 和 deadline。
+原生交接先接收并读取一条有界请求，再校验捕获根目录的物理路径。目录缺失或为
+符号链接时关闭该连接，原来源可纠正路径后经同一监听器继续；拒绝不捕获、不释放
+旧 packet，不延长原 wake 截止时间，也不创建或跟随外来根目录。
+连接尚未接收时的 Host 故障仍是执行失败，不是一次 packet 拒绝，不进入保留
+来源的请求重试循环。
 显式开启的两处完整 controller 捕获 hook 共用 90 秒窗口（原为 30 秒），覆盖完整原生/研究上下文
 准备、一致性复核与 live-source 捕获，并受原试验 phase 截止时间约束。controller
 沿用同一个绝对截止时间；解除暂停后，只能在原 phase 内再用至多 30 秒封存静态副本。

benchmark/runtime/checkpoint.py

@@ -415,7 +415,7 @@ def capture_after_wake(env: dict[str, str], receipt: dict[str, Any], *,
 @contextmanager
 def held_quota_decision(listener, capture_root: Path, *, goal_id: str, agent_id: str,
                         turn_instance_id: str | None, native_process=None, interrupt_timeout: float = 10,
-                        interrupt_when=None, deadline: float | None = None):
+                        interrupt_when=None, deadline: float | None = None, accepted_channel=None):
     """Host-side counterpart of the explicit quota pre-output transport.
 
     The existing typed owner decides whether this exact packet admits a fresh
@@ -436,19 +436,24 @@ def held_quota_decision(listener, capture_root: Path, *, goal_id: str, agent_id:
     Without native_process this is output ordering only; a slow capture can
     outlive the native tool's yield interval and does not stop inference.
     """
-    if not capture_root.is_absolute() or capture_root.resolve(strict=True) != capture_root:
-        raise ValueError("decision capture root must be a physical absolute directory")
     def remaining():
         limit = 120 if deadline is None else min(120, deadline - time.monotonic())
         if limit <= 0:
             raise TimeoutError("original wake budget expired during decision handoff")
         return limit
 
-    channel, _ = listener.accept()
+    channel = accepted_channel
+    if channel is None:
+        channel, _ = listener.accept()
     with channel:
         channel.settimeout(remaining())
         with channel.makefile("rb") as stream:
             line = stream.readline(4097)
+        # Consume the bounded request before checking Host prerequisites. A
+        # refused root must close this channel so the same source can recover;
+        # leaving it queued would keep the caller's listener always readable.
+        if not capture_root.is_absolute() or capture_root.resolve(strict=True) != capture_root:
+            raise ValueError("decision capture root must be a physical absolute directory")
         message = json.loads(line)
         if (not isinstance(message, dict) or set(message) != {"path", "sha256"} or
                 not isinstance(message["path"], str) or not isinstance(message["sha256"], str)):

benchmark/runtime/checkpoint_wake.py

@@ -203,25 +203,42 @@ def run(self, execution, body, wake, session_wake, stdout, stderr, timeout):
                         raise subprocess.TimeoutExpired(command, timeout)
                     if not select.select([self.listener], [], [], min(.1, remaining))[0]:
                         continue
-                    with held_quota_decision(self.listener, self.capture_root,
-                            goal_id=self.env["LOOPX_GOAL_ID"], agent_id=self.env["LOOPX_AGENT_ID"],
-                            turn_instance_id=None, native_process=process,
-                            interrupt_timeout=min(10, remaining), interrupt_when=self._first_replan,
-                            deadline=deadline) as held:
-                        if "native_process_return_code" not in held:
-                            continue
-                        stdout.flush()
-                        session_wake.observe(wake / "stdout.jsonl")
-                        event = {"obligation_id": held["intent"]["obligation_id"],
-                                 "decision_path": str(held["path"]), "decision_sha256": held["sha256"],
-                                 "captured": False, "research_point_qualified": False}
-                        self.receipt.setdefault("checkpoint_handoffs", []).append(event)
+                    # Only a consumed request can be refused and retried.
+                    # Accept failures belong to the Host, not the packet.
+                    with self.listener.accept()[0] as channel:
+                        entered = False
                         try:
-                            event.update(self._capture(held, deadline), captured=True)
+                            with held_quota_decision(self.listener, self.capture_root,
+                                    goal_id=self.env["LOOPX_GOAL_ID"], agent_id=self.env["LOOPX_AGENT_ID"],
+                                    turn_instance_id=None, native_process=process,
+                                    interrupt_timeout=min(10, remaining), interrupt_when=self._first_replan,
+                                    deadline=deadline, accepted_channel=channel) as held:
+                                entered = True
+                                if "native_process_return_code" not in held:
+                                    continue
+                                stdout.flush()
+                                session_wake.observe(wake / "stdout.jsonl")
+                                event = {"obligation_id": held["intent"]["obligation_id"],
+                                         "decision_path": str(held["path"]), "decision_sha256": held["sha256"],
+                                         "captured": False, "research_point_qualified": False}
+                                self.receipt.setdefault("checkpoint_handoffs", []).append(event)
+                                try:
+                                    event.update(self._capture(held, deadline), captured=True)
+                                except Exception as error:
+                                    # A missed optional point cannot discard the original
+                                    # committed guard or turn a later retry into its capture.
+                                    event.update(error_kind=type(error).__name__, error=str(error)[:500])
                         except Exception as error:
-                            # A missed optional point cannot discard the original
-                            # committed guard or turn a later retry into its capture.
-                            event.update(error_kind=type(error).__name__, error=str(error)[:500])
+                            # A refused handoff (unparseable, outside this wake's capture
+                            # root, or rejected by the typed selector) is neither released
+                            # nor captured; its caller already reports a typed transport
+                            # failure. The uninterrupted original wake keeps running.
+                            if entered or process.poll() is not None:
+                                raise
+                            self.receipt.setdefault("checkpoint_handoffs", []).append({
+                                "captured": False, "research_point_qualified": False, "released": False,
+                                "error_kind": type(error).__name__, "error": str(error)[:500]})
+                            continue
                     continuation = held.get("continuation_body")
                     if continuation is not None:
                         break

benchmark/runtime/planning.py

@@ -8,8 +8,25 @@
 from pathlib import Path
 
 
+_RESEARCH_PLANNING_GUIDANCE = (
+    "Before an expensive implementation, check the declared interface, units, scale "
+    "and other decision-critical assumptions against the available task material. "
+    "Compare a feasible alternative with the strongest available starting point; "
+    "preserve that reference while testing a replacement. Use a cheap discriminating "
+    "check when it can resolve the choice, then budget enough work for the selected "
+    "complete experiment. An informative negative experiment can complete its own "
+    "Todo without satisfying the broader goal. Keep evaluation readiness separate "
+    "from evidence for replacing the best artifact; neither creates a new acceptance "
+    "document or permission requirement."
+)
+
+
 def render_planning_prompt(packet: dict, env: dict[str, str]) -> str:
-    """Preserve the task packet without projecting the Host's execution clock."""
+    """Compose benchmark-only research guidance without changing the Core packet.
+
+    The worker calls this only for its explicit loopx-planned task entry. Core
+    Goal/CLI planning and plain native benchmark workers do not use this wrapper.
+    """
     if "LOOPX_PLANNING_TIMEOUT_SEC" in env:
         raise ValueError(
             "Remove LOOPX_PLANNING_TIMEOUT_SEC; planning uses LOOPX_PHASE_DEADLINE_EPOCH"
@@ -19,7 +36,8 @@ def render_planning_prompt(packet: dict, env: dict[str, str]) -> str:
         value = float(env[key])
         if not math.isfinite(value) or value <= 0:
             raise ValueError(f"{key} must be finite and positive")
-    return "$loopx\n\nHost-supplied planning checkpoint:\n" + json.dumps(packet, ensure_ascii=False)
+    return ("$loopx\n\nBenchmark research planning guidance:\n" + _RESEARCH_PLANNING_GUIDANCE
+            + "\n\nHost-supplied planning checkpoint:\n" + json.dumps(packet, ensure_ascii=False))
 
 
 def task_plan_packet(env: dict[str, str], cli: list[str]) -> dict:

benchmark/runtime/sforge.py

@@ -355,6 +355,10 @@ def format_run_cmd(self, prompt_path, *, model=None, cwd="", internet=True, resu
             with tempfile.TemporaryDirectory(prefix="benchmark-prompt-") as directory:
                 prompt = Path(directory) / "task.md"
                 prompt.write_text(self.feedback_prompt)
+                # The native copy preserves the host mode and lands root-owned.
+                # The container's agent user must read the prompt whatever
+                # umask the launcher runs under.
+                prompt.chmod(0o644)
                 asyncio.run(self.environment.upload_file(prompt, prompt_path))
             # Native run_agent saved the pre-override wrapper. Keep its visible
             # prompt artifact aligned with what all five workers actually read.

benchmark/runtime/sforge_entry.py

@@ -13,6 +13,17 @@
 from .planning import task_plan_packet, validate_plan_readback
 from .worker import PHASE_WORK_RESERVE_SECONDS, loopx_command, run_once
 
+# One planning wake whose native process exits non-zero without a harness
+# error or a validated plan (a dropped model stream, a failed turn) must not
+# strand the whole trial before execution. Retry within the same phase budget.
+PLANNING_WAKE_ATTEMPTS = 3
+
+
+def _retryable_planning_wake(receipt: dict) -> bool:
+    return (receipt.get("ok") is not True and receipt.get("timed_out") is not True
+            and "error_kind" not in receipt and "planning" not in receipt
+            and receipt.get("return_code") not in (None, 0))
+
 
 def prepare_entry(env: dict[str, str]) -> bool:
     """Reuse a qualified initial plan; execution still requires ordinary quota."""
@@ -48,7 +59,17 @@ def prepare_entry(env: dict[str, str]) -> bool:
     # Native SForge owns one absolute trial deadline for planning and execution.
     # Do not inherit a separate planning cap from the shared worker environment.
     planning_env.pop("LOOPX_PLANNING_TIMEOUT_SEC", None)
-    receipt = run_once(planning_env)
+    attempts = []
+    for attempt in range(1, PLANNING_WAKE_ATTEMPTS + 1):
+        receipt = run_once(planning_env)
+        attempts.append({"attempt": attempt, "ok": receipt.get("ok") is True,
+                         "timed_out": receipt.get("timed_out") is True,
+                         "return_code": receipt.get("return_code"),
+                         **({"error_kind": receipt["error_kind"]} if "error_kind" in receipt else {})})
+        remaining = float(env["LOOPX_PHASE_DEADLINE_EPOCH"]) - time.time()
+        if not _retryable_planning_wake(receipt) or remaining <= PHASE_WORK_RESERVE_SECONDS:
+            break
+    path.with_name("planning-attempts.json").write_text(json.dumps(attempts))
     if receipt.get("ok") is not True:
         raise RuntimeError("Planning failed; execution was not started")
     entry = receipt.get("planning") or {}

benchmark/runtime/worker.py

@@ -329,6 +329,11 @@ class _InertCheckpointWake:
     listener = None
 
 
+def _error_tail(exc: BaseException) -> dict[str, str]:
+    """Private wake evidence: the class plus a bounded message, never a traceback."""
+    return {"error_kind": type(exc).__name__, "error": " ".join(str(exc).split())[:500]}
+
+
 def run_once(env: dict[str, str]) -> dict:
     if env.get("LOOPX_TURN_ENVELOPE", "0") not in {"0", "1"}:
         raise ValueError("LOOPX_TURN_ENVELOPE must be 0 or 1")
@@ -501,15 +506,15 @@ def run_once(env: dict[str, str]) -> dict:
         receipt["timed_out"] = True
     except BaseException as exc:
         receipt["ok"] = False
-        receipt["error_kind"] = type(exc).__name__
+        receipt.update(_error_tail(exc))
         raise
     finally:
         checkpoint_context.close()
         try:
             if session_wake is not None:
                 session_wake.observe(wake / "stdout.jsonl")
         except BaseException as exc:
-            receipt.update(ok=False, error_kind=type(exc).__name__)
+            receipt.update(ok=False, **_error_tail(exc))
             raise
         finally:
             if env.get("LOOPX_CHECKPOINT_REQUEST"):