benchmark: integrate recovery fixes for continued EdgeBench work

benchmark/README.md

@@ -708,6 +708,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
@@ -924,6 +932,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/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"):

benchmark/tests/test_checkpoint_handoff.py

@@ -482,3 +482,135 @@ def gate(*args, full_history):
     assert checkpoint.capture_before_replan(env, receipt, "current work").endswith("current work")
     assert observed == [not native_participant]
     assert receipt["checkpoint"]["reason"] == "no_fresh_replan"
+
+
+@pytest.mark.skipif(os.name != "posix", reason="Requires the Host Unix rendezvous")
+@pytest.mark.parametrize("capture_root_state", ["present", "missing", "symlink"])
+def test_refused_handoff_keeps_original_wake_running(tmp_path, monkeypatch, capture_root_state):
+    """A resumed session reusing an earlier wake's capture root is refused, not fatal.
+
+    The refused packet is neither released nor captured; the same source keeps
+    using the private endpoint and exits on its own. The receipt keeps a bounded
+    reason instead of only an exception class.
+    """
+    import socket
+    import tempfile
+    from pathlib import Path
+    from types import SimpleNamespace
+
+    import benchmark.runtime.worker as worker
+    from benchmark.runtime.checkpoint_wake import CheckpointWake
+
+    wakes = tmp_path / "wakes"
+    current = wakes / "wake-current" / "decisions"
+    earlier = wakes / "wake-earlier" / "decisions" / "decision-earlier"
+    current.parent.mkdir(parents=True)
+    if capture_root_state == "present":
+        current.mkdir(mode=0o700)
+    foreign = tmp_path / "foreign"
+    foreign.mkdir()
+    if capture_root_state == "symlink":
+        current.symlink_to(foreign, target_is_directory=True)
+    earlier.mkdir(parents=True)
+    released = current / "decision-current"
+    packets = {}
+    for name, directory in (("earlier", earlier), ("current", released)):
+        raw = json.dumps({"ok": False, "goal_id": "source", "error_code": "fixture"}).encode()
+        if name == "earlier":
+            (directory / "decision.json").write_bytes(raw)
+        packets[name] = {"path": str(directory / "decision.json"), "sha256": hashlib.sha256(raw).hexdigest()}
+    project = tmp_path / "project"
+    project.mkdir()
+    env = {**os.environ, "LOOPX_WAKE_LOG_DIR": str(wakes), "LOOPX_GOAL_ID": "source",
+           "LOOPX_AGENT_ID": "agent", "LOOPX_PROJECT": str(project)}
+    with tempfile.TemporaryDirectory(prefix="refused-", dir="/tmp") as short:
+        address = str(Path(short).resolve() / "host.sock")
+        code = (
+            "import json,socket,sys; from pathlib import Path\n"
+            "def call(message):\n"
+            "    with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as s:\n"
+            f"        s.connect({address!r}); s.sendall(json.dumps(message).encode()+b'\\n')\n"
+            "        return s.makefile('rb').readline()\n"
+            f"assert call({packets['earlier']!r}) == b''\n"
+            f"current = Path({str(current)!r})\n"
+            "if current.is_symlink(): current.unlink()\n"
+            f"Path({str(released)!r}).mkdir(parents=True)\n"
+            f"Path({packets['current']['path']!r}).write_bytes({raw!r})\n"
+            f"assert json.loads(call({packets['current']!r}))['released'] is True\n"
+            "Path('continued').write_text('source continued')\n")
+        monkeypatch.setattr(worker, "native_command", lambda *args: [sys.executable, "-c", code])
+        wake = CheckpointWake(env, {"turn_id": "wake-current", "mode": "heartbeat", "stage": "execute"})
+        wake.capture_root = current
+        with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as listener:
+            listener.bind(address)
+            listener.listen(1)
+            wake.listener = listener
+            with (tmp_path / "stdout").open("w") as stdout, (tmp_path / "stderr").open("w") as stderr:
+                assert wake.run(SimpleNamespace(), "body", tmp_path, None, stdout, stderr, 2) == 0
+    assert (project / "continued").read_text() == "source continued"
+    [event] = wake.receipt["checkpoint_handoffs"]
+    assert event["captured"] is False and event["released"] is False
+    expected = "FileNotFoundError" if capture_root_state == "missing" else "ValueError"
+    assert event["error_kind"] == expected
+    assert (earlier / "decision.json").read_bytes() == raw
+    assert not list(foreign.iterdir())  # Refusal cannot write through a foreign root.
+
+
+def test_wake_receipt_error_tail_is_bounded_message():
+    from benchmark.runtime.worker import _error_tail
+
+    tail = _error_tail(ValueError("decision handoff must reference\nthis Host's capture root " + "x" * 900))
+    assert tail["error_kind"] == "ValueError"
+    assert tail["error"].startswith("decision handoff must reference this Host's capture root")
+    assert len(tail["error"]) == 500 and "\n" not in tail["error"]
+
+
+@pytest.mark.skipif(os.name != "posix", reason="Requires the Host Unix rendezvous")
+def test_accept_failure_is_not_retried_as_a_refused_packet(tmp_path, monkeypatch):
+    import errno
+    import socket
+    import tempfile
+    from contextlib import contextmanager
+    from pathlib import Path
+    from types import SimpleNamespace
+
+    import benchmark.runtime.worker as worker
+    from benchmark.runtime.checkpoint_wake import CheckpointWake
+
+    processes, accepts = [], []
+    owned_child = worker.child_process
+
+    @contextmanager
+    def record_child(*args, **kwargs):
+        with owned_child(*args, **kwargs) as process:
+            processes.append(process)
+            yield process
+
+    monkeypatch.setattr(worker, "child_process", record_child)
+    env = {**os.environ, "LOOPX_PROJECT": str(tmp_path), "LOOPX_WAKE_LOG_DIR": str(tmp_path),
+           "LOOPX_GOAL_ID": "source", "LOOPX_AGENT_ID": "agent"}
+    wake = CheckpointWake(env, {"turn_id": "turn", "mode": "heartbeat", "stage": "execute"})
+    with tempfile.TemporaryDirectory(prefix="accept-failure-", dir="/tmp") as short:
+        address = str(Path(short).resolve() / "host.sock")
+        with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as listener:
+            listener.bind(address)
+            listener.listen(1)
+
+            class UnavailableListener:
+                def fileno(self):
+                    return listener.fileno()
+
+                def accept(self):
+                    accepts.append(True)
+                    raise OSError(errno.EMFILE, "file descriptor capacity exhausted")
+
+            wake.listener = UnavailableListener()
+            code = ("import socket; s=socket.socket(socket.AF_UNIX,socket.SOCK_STREAM); "
+                    f"s.connect({address!r}); s.sendall(b'{{}}\\n'); s.recv(1)")
+            monkeypatch.setattr(worker, "native_command", lambda *args: [sys.executable, "-c", code])
+            with (tmp_path / "out").open("w") as stdout, (tmp_path / "err").open("w") as stderr:
+                with pytest.raises(OSError, match="capacity exhausted"):
+                    wake.run(SimpleNamespace(), "body", tmp_path, None, stdout, stderr, 1)
+    assert len(accepts) == 1
+    assert "checkpoint_handoffs" not in wake.receipt
+    assert len(processes) == 1 and processes[0].poll() is not None