Skip to content

Commit f9bdae6

Browse files
authored
Merge pull request #6624 from loopx-project/edgebenchtest
benchmark: integrate recovery fixes for continued EdgeBench work
2 parents 6c0cfbf + 8d89fa5 commit f9bdae6

25 files changed

Lines changed: 1188 additions & 57 deletions

‎benchmark/README.md‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -719,6 +719,14 @@ Omit the controller file and `controller_handoff` on a new run to disable automa
719719
controller capture; removing the checkpoint request disables capture entirely.
720720
Existing artifacts remain inert, and the source trial is not restarted by disabling it.
721721

722+
The native handoff accepts and reads one bounded request before validating its
723+
physical capture root. A missing or symlinked root closes that connection;
724+
the original source can correct its path and continue through the same listener.
725+
The refusal neither captures nor releases the old packet and does not renew the
726+
original wake deadline. The transport does not create or follow a foreign root.
727+
Host failures before a connection is accepted remain execution failures. They
728+
are not packet refusals and cannot enter the source-preserving retry loop.
729+
722730
To reproduce **joint source capture and both fresh-guest receivers**, use
723731
[`qualify_checkpoint_joint.py`](tests/qualify_checkpoint_joint.py) from the current
724732
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
831839
prevents new planning-clock injection; it does not qualify old-archive recovery
832840
or claim improved model behavior or scores.
833841

842+
Only the explicit `loopx-planned` benchmark worker adds research-planning
843+
guidance around the unchanged product planning packet: check decision-critical
844+
assumptions, compare with and retain the strongest available reference, and keep
845+
an informative negative experiment separate from Goal completion. This is a
846+
provider planning treatment, not new default Core/CLI guidance or a checkpoint
847+
requirement. Ordinary Goal entry, seeded work and official native workers do not
848+
use this wrapper. Pin this provider revision when comparing runs; the guidance
849+
does not establish model adoption, initial-quality parity or long-horizon gain.
850+
851+
仅显式 `loopx-planned` benchmark worker 在未修改的产品规划 packet 外附加研究
852+
指引:核对关键假设、对照并保留最强起点、区分有信息量的负实验与 Goal 完成。
853+
这是 provider 规划处理,不是 Core/CLI 默认指引或 checkpoint 要求。普通 Goal、
854+
预置任务和 official 原生 worker 不使用此包装;对比运行须固定 provider revision。
855+
这些指引不证明模型采用、首次质量持平或长程收益。
856+
834857
The sealed deadline and native capture timestamp determine the remainder; the
835858
receiver recomputes it. A point with two hours remaining keeps that duration
836859
after an eight-hour offline gap. It does not reuse the old absolute epoch or
@@ -935,6 +958,11 @@ worker 保留原 guard,controller 暂停全部容器写入者,核对仍有
935958
文件 witness,再封存原生评测机会、incumbent 及剩余预算。源变化、过期或失去
936959
交接会拒绝;可恢复的捕获失败不取消原研究,也不得改用稍后的状态。解除暂停后,
937960
源会话继续使用原 packet 和 deadline。
961+
原生交接先接收并读取一条有界请求,再校验捕获根目录的物理路径。目录缺失或为
962+
符号链接时关闭该连接,原来源可纠正路径后经同一监听器继续;拒绝不捕获、不释放
963+
旧 packet,不延长原 wake 截止时间,也不创建或跟随外来根目录。
964+
连接尚未接收时的 Host 故障仍是执行失败,不是一次 packet 拒绝,不进入保留
965+
来源的请求重试循环。
938966
显式开启的两处完整 controller 捕获 hook 共用 90 秒窗口(原为 30 秒),覆盖完整原生/研究上下文
939967
准备、一致性复核与 live-source 捕获,并受原试验 phase 截止时间约束。controller
940968
沿用同一个绝对截止时间;解除暂停后,只能在原 phase 内再用至多 30 秒封存静态副本。

‎benchmark/runtime/checkpoint.py‎

Lines changed: 9 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -415,7 +415,7 @@ def capture_after_wake(env: dict[str, str], receipt: dict[str, Any], *,
415415
@contextmanager
416416
def held_quota_decision(listener, capture_root: Path, *, goal_id: str, agent_id: str,
417417
turn_instance_id: str | None, native_process=None, interrupt_timeout: float = 10,
418-
interrupt_when=None, deadline: float | None = None):
418+
interrupt_when=None, deadline: float | None = None, accepted_channel=None):
419419
"""Host-side counterpart of the explicit quota pre-output transport.
420420
421421
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:
436436
Without native_process this is output ordering only; a slow capture can
437437
outlive the native tool's yield interval and does not stop inference.
438438
"""
439-
if not capture_root.is_absolute() or capture_root.resolve(strict=True) != capture_root:
440-
raise ValueError("decision capture root must be a physical absolute directory")
441439
def remaining():
442440
limit = 120 if deadline is None else min(120, deadline - time.monotonic())
443441
if limit <= 0:
444442
raise TimeoutError("original wake budget expired during decision handoff")
445443
return limit
446444

447-
channel, _ = listener.accept()
445+
channel = accepted_channel
446+
if channel is None:
447+
channel, _ = listener.accept()
448448
with channel:
449449
channel.settimeout(remaining())
450450
with channel.makefile("rb") as stream:
451451
line = stream.readline(4097)
452+
# Consume the bounded request before checking Host prerequisites. A
453+
# refused root must close this channel so the same source can recover;
454+
# leaving it queued would keep the caller's listener always readable.
455+
if not capture_root.is_absolute() or capture_root.resolve(strict=True) != capture_root:
456+
raise ValueError("decision capture root must be a physical absolute directory")
452457
message = json.loads(line)
453458
if (not isinstance(message, dict) or set(message) != {"path", "sha256"} or
454459
not isinstance(message["path"], str) or not isinstance(message["sha256"], str)):

‎benchmark/runtime/checkpoint_wake.py‎

Lines changed: 34 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -203,25 +203,42 @@ def run(self, execution, body, wake, session_wake, stdout, stderr, timeout):
203203
raise subprocess.TimeoutExpired(command, timeout)
204204
if not select.select([self.listener], [], [], min(.1, remaining))[0]:
205205
continue
206-
with held_quota_decision(self.listener, self.capture_root,
207-
goal_id=self.env["LOOPX_GOAL_ID"], agent_id=self.env["LOOPX_AGENT_ID"],
208-
turn_instance_id=None, native_process=process,
209-
interrupt_timeout=min(10, remaining), interrupt_when=self._first_replan,
210-
deadline=deadline) as held:
211-
if "native_process_return_code" not in held:
212-
continue
213-
stdout.flush()
214-
session_wake.observe(wake / "stdout.jsonl")
215-
event = {"obligation_id": held["intent"]["obligation_id"],
216-
"decision_path": str(held["path"]), "decision_sha256": held["sha256"],
217-
"captured": False, "research_point_qualified": False}
218-
self.receipt.setdefault("checkpoint_handoffs", []).append(event)
206+
# Only a consumed request can be refused and retried.
207+
# Accept failures belong to the Host, not the packet.
208+
with self.listener.accept()[0] as channel:
209+
entered = False
219210
try:
220-
event.update(self._capture(held, deadline), captured=True)
211+
with held_quota_decision(self.listener, self.capture_root,
212+
goal_id=self.env["LOOPX_GOAL_ID"], agent_id=self.env["LOOPX_AGENT_ID"],
213+
turn_instance_id=None, native_process=process,
214+
interrupt_timeout=min(10, remaining), interrupt_when=self._first_replan,
215+
deadline=deadline, accepted_channel=channel) as held:
216+
entered = True
217+
if "native_process_return_code" not in held:
218+
continue
219+
stdout.flush()
220+
session_wake.observe(wake / "stdout.jsonl")
221+
event = {"obligation_id": held["intent"]["obligation_id"],
222+
"decision_path": str(held["path"]), "decision_sha256": held["sha256"],
223+
"captured": False, "research_point_qualified": False}
224+
self.receipt.setdefault("checkpoint_handoffs", []).append(event)
225+
try:
226+
event.update(self._capture(held, deadline), captured=True)
227+
except Exception as error:
228+
# A missed optional point cannot discard the original
229+
# committed guard or turn a later retry into its capture.
230+
event.update(error_kind=type(error).__name__, error=str(error)[:500])
221231
except Exception as error:
222-
# A missed optional point cannot discard the original
223-
# committed guard or turn a later retry into its capture.
224-
event.update(error_kind=type(error).__name__, error=str(error)[:500])
232+
# A refused handoff (unparseable, outside this wake's capture
233+
# root, or rejected by the typed selector) is neither released
234+
# nor captured; its caller already reports a typed transport
235+
# failure. The uninterrupted original wake keeps running.
236+
if entered or process.poll() is not None:
237+
raise
238+
self.receipt.setdefault("checkpoint_handoffs", []).append({
239+
"captured": False, "research_point_qualified": False, "released": False,
240+
"error_kind": type(error).__name__, "error": str(error)[:500]})
241+
continue
225242
continuation = held.get("continuation_body")
226243
if continuation is not None:
227244
break

‎benchmark/runtime/planning.py‎

Lines changed: 20 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -8,8 +8,25 @@
88
from pathlib import Path
99

1010

11+
_RESEARCH_PLANNING_GUIDANCE = (
12+
"Before an expensive implementation, check the declared interface, units, scale "
13+
"and other decision-critical assumptions against the available task material. "
14+
"Compare a feasible alternative with the strongest available starting point; "
15+
"preserve that reference while testing a replacement. Use a cheap discriminating "
16+
"check when it can resolve the choice, then budget enough work for the selected "
17+
"complete experiment. An informative negative experiment can complete its own "
18+
"Todo without satisfying the broader goal. Keep evaluation readiness separate "
19+
"from evidence for replacing the best artifact; neither creates a new acceptance "
20+
"document or permission requirement."
21+
)
22+
23+
1124
def render_planning_prompt(packet: dict, env: dict[str, str]) -> str:
12-
"""Preserve the task packet without projecting the Host's execution clock."""
25+
"""Compose benchmark-only research guidance without changing the Core packet.
26+
27+
The worker calls this only for its explicit loopx-planned task entry. Core
28+
Goal/CLI planning and plain native benchmark workers do not use this wrapper.
29+
"""
1330
if "LOOPX_PLANNING_TIMEOUT_SEC" in env:
1431
raise ValueError(
1532
"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:
1936
value = float(env[key])
2037
if not math.isfinite(value) or value <= 0:
2138
raise ValueError(f"{key} must be finite and positive")
22-
return "$loopx\n\nHost-supplied planning checkpoint:\n" + json.dumps(packet, ensure_ascii=False)
39+
return ("$loopx\n\nBenchmark research planning guidance:\n" + _RESEARCH_PLANNING_GUIDANCE
40+
+ "\n\nHost-supplied planning checkpoint:\n" + json.dumps(packet, ensure_ascii=False))
2341

2442

2543
def task_plan_packet(env: dict[str, str], cli: list[str]) -> dict:

‎benchmark/runtime/sforge.py‎

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

‎benchmark/runtime/sforge_entry.py‎

Lines changed: 22 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,17 @@
1313
from .planning import task_plan_packet, validate_plan_readback
1414
from .worker import PHASE_WORK_RESERVE_SECONDS, loopx_command, run_once
1515

16+
# One planning wake whose native process exits non-zero without a harness
17+
# error or a validated plan (a dropped model stream, a failed turn) must not
18+
# strand the whole trial before execution. Retry within the same phase budget.
19+
PLANNING_WAKE_ATTEMPTS = 3
20+
21+
22+
def _retryable_planning_wake(receipt: dict) -> bool:
23+
return (receipt.get("ok") is not True and receipt.get("timed_out") is not True
24+
and "error_kind" not in receipt and "planning" not in receipt
25+
and receipt.get("return_code") not in (None, 0))
26+
1627

1728
def prepare_entry(env: dict[str, str]) -> bool:
1829
"""Reuse a qualified initial plan; execution still requires ordinary quota."""
@@ -48,7 +59,17 @@ def prepare_entry(env: dict[str, str]) -> bool:
4859
# Native SForge owns one absolute trial deadline for planning and execution.
4960
# Do not inherit a separate planning cap from the shared worker environment.
5061
planning_env.pop("LOOPX_PLANNING_TIMEOUT_SEC", None)
51-
receipt = run_once(planning_env)
62+
attempts = []
63+
for attempt in range(1, PLANNING_WAKE_ATTEMPTS + 1):
64+
receipt = run_once(planning_env)
65+
attempts.append({"attempt": attempt, "ok": receipt.get("ok") is True,
66+
"timed_out": receipt.get("timed_out") is True,
67+
"return_code": receipt.get("return_code"),
68+
**({"error_kind": receipt["error_kind"]} if "error_kind" in receipt else {})})
69+
remaining = float(env["LOOPX_PHASE_DEADLINE_EPOCH"]) - time.time()
70+
if not _retryable_planning_wake(receipt) or remaining <= PHASE_WORK_RESERVE_SECONDS:
71+
break
72+
path.with_name("planning-attempts.json").write_text(json.dumps(attempts))
5273
if receipt.get("ok") is not True:
5374
raise RuntimeError("Planning failed; execution was not started")
5475
entry = receipt.get("planning") or {}

‎benchmark/runtime/worker.py‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -329,6 +329,11 @@ class _InertCheckpointWake:
329329
listener = None
330330

331331

332+
def _error_tail(exc: BaseException) -> dict[str, str]:
333+
"""Private wake evidence: the class plus a bounded message, never a traceback."""
334+
return {"error_kind": type(exc).__name__, "error": " ".join(str(exc).split())[:500]}
335+
336+
332337
def run_once(env: dict[str, str]) -> dict:
333338
if env.get("LOOPX_TURN_ENVELOPE", "0") not in {"0", "1"}:
334339
raise ValueError("LOOPX_TURN_ENVELOPE must be 0 or 1")
@@ -501,15 +506,15 @@ def run_once(env: dict[str, str]) -> dict:
501506
receipt["timed_out"] = True
502507
except BaseException as exc:
503508
receipt["ok"] = False
504-
receipt["error_kind"] = type(exc).__name__
509+
receipt.update(_error_tail(exc))
505510
raise
506511
finally:
507512
checkpoint_context.close()
508513
try:
509514
if session_wake is not None:
510515
session_wake.observe(wake / "stdout.jsonl")
511516
except BaseException as exc:
512-
receipt.update(ok=False, error_kind=type(exc).__name__)
517+
receipt.update(ok=False, **_error_tail(exc))
513518
raise
514519
finally:
515520
if env.get("LOOPX_CHECKPOINT_REQUEST"):

0 commit comments

Comments
 (0)