Share Explore content capture and independent native artifact evidence

benchmark/edgebench/README.md

@@ -429,6 +429,47 @@ resource policy in the study manifest before launching a new comparison.
 Use explicit `--feedback best-only` to retain strict-improvement publication, or
 `--feedback blind` as an evaluator-feedback-free control. Harbor is unchanged.
 
+### Native artifact evidence and independent measurements
+
+Native heartbeat workers with `--execution-observation local_probe` expose
+`loopx-bind-native-candidate` as an artifact reader. Native results do not need
+a preceding successful local execution:
+
+```sh
+loopx-bind-native-candidate --result native-result.json
+```
+
+The original receipt names the accepted upload as `archive:<sha256>` and records
+its size. This offline command checks the receipt shape, not service identity.
+For additional file-content evidence, pass `--archive submission.tar.gz`. It
+checks the original receipt against the bytes and safely reads the complete
+ordinary gzip tar without extracting files, bounded to 256 members and 64 MiB
+compressed/expanded. The returned `submitted_file_set_ref` uses the shared
+Explore `candidate_content` path-and-byte manifest. File mode, packing, build
+lineage and execution conditions remain separate; equal content refs alone do
+not prove equivalent execution. Generated outputs belong to the full file set,
+and a probe's selected subset or extra fixtures are a different subject.
+
+Local and native producers each retain their own actual workload, evaluator,
+subject and evidence through ordinary Explore measurements. There is no
+required local-to-native work order. Missing content evidence limits pairing,
+not retention of an otherwise valid result. Never rewrite an earlier
+measurement or resubmit merely to force a candidate match. When the installed
+native client advertises `--retain-archive` and `--json`, it can retain bytes of
+a planned upload; this is optional evidence, not an extra evaluation.
+
+The old `--observation`, `--snapshot` and `--generated-file` options remain an
+explicit offline diagnostic for sealed probe artifacts. They keep their strict
+original correspondence checks and do not gate ordinary native facts.
+
+The common content capture lives in the product's existing Explore producer
+neighborhood; benchmark runtime only adapts native receipt/archive fields. A
+split product/runner install selecting capture must pin a product revision with
+`loopx.capabilities.explore.candidate_content`. Preflight rejects an older
+product before worker staging; it never overlays product code or falls back to
+a benchmark copy. Omitted capture preserves the historical source-pin behavior.
+
+
 For native archive submission, start `python -m benchmark.edgebench.serve` and use the runner
 without `--feedback` to select the defaults. Explicit
 `--feedback native --submission-cooldown 120 --eval-interval 300` restores the

benchmark/edgebench/prompts.py

@@ -17,7 +17,7 @@ def native_research_guidance() -> str:
     )
 
 
-def research_artifact_instructions(*, checkpoint: bool) -> str:
+def research_artifact_instructions(*, checkpoint: bool, native_probe: bool = False) -> str:
     """Separate reproducibility data from candidate inputs for every native arm."""
     from benchmark.runtime.research_artifacts import RESEARCH_DIRECTORY
 
@@ -28,6 +28,16 @@ def research_artifact_instructions(*, checkpoint: bool) -> str:
         "Terminal artifact collection retains this directory for analysis; "
         "that alone does not provide a resumable checkpoint. "
     )
+    binding = (
+        "For native artifact evidence, retain the planned upload with sforge-submit --retain-archive "
+        "when the installed client advertises it, then read the receipt and archive using "
+        "loopx-bind-native-candidate --help. This does not require a local probe or a new submission. "
+        "Keep archive identity and file-content identity distinct: content refs do not cover modes, "
+        "packaging or execution conditions. Record available measurements through ordinary Explore "
+        "writeback with each producer's actual workload and evaluator; missing or non-equivalent "
+        "candidate evidence limits pairing, not retention of either result. Preserve earlier refs.\n\n"
+        if native_probe else ""
+    )
     return (
         "### Research Artifacts\n\n"
         f"Use `{RESEARCH_DIRECTORY}` for reusable research artifacts that are not "
@@ -42,6 +52,7 @@ def research_artifact_instructions(*, checkpoint: bool) -> str:
         "explicit until they have been copied and their references updated and checked. "
         "Older checkpoints preserve their original workspace files; this policy does "
         "not relocate or discard historical evidence.\n\n"
+        f"{binding}"
         "This storage guidance does not change submitted files, task acceptance, "
         "feedback access or execution authority. Capture and receiver verification "
         "still determine which artifacts are actually available.\n\n"

benchmark/edgebench/run.py

@@ -322,6 +322,10 @@ def main(argv=None):
     runner = Path(os.environ.get("LOOPX_RUNNER_SRC_DIR", str(product))).resolve()
     pins = source_pins(product, runner, os.environ["LOOPX_EXPECTED_COMMIT"],
                        os.environ.get("LOOPX_EXPECTED_RUNNER_COMMIT"))
+    if args.execution_observation is not None:
+        from benchmark.runtime.source import require_candidate_content
+
+        require_candidate_content(product, pins[0])
     if args.restore_checkpoint is not None:
         # Activating the native capsule starts its remaining phase clock.
         # Qualify the actual install archives first, including ignored assets;
@@ -367,7 +371,10 @@ def main(argv=None):
             eval_interval=args.eval_interval, internet=task.internet,
             selection=task.judge.selection, score_direction=task.judge.score_direction)
         workspace_instructions = native_task_instructions(task.submit_paths, **native_policy)
-        workspace_instructions += research_artifact_instructions(checkpoint=checkpoint_request is not None)
+        workspace_instructions += research_artifact_instructions(
+            checkpoint=checkpoint_request is not None,
+            native_probe=args.execution_observation is not None,
+        )
         # All native arms get the same storage policy. Checkpoint coverage is
         # explicit; neither collection nor a sidecar changes native submissions.
         feedback_prompt = workspace_instructions + f"---\n\n{task.work.agent_query}\n"

benchmark/runtime/harbor.py

@@ -228,6 +228,10 @@ async def _stage_source(self, environment: BaseEnvironment, source: Path) -> str
             source, runner, os.environ.get("LOOPX_EXPECTED_COMMIT"),
             os.environ.get("LOOPX_EXPECTED_RUNNER_COMMIT"),
         )
+        if self.execution_observation is not None:
+            from .source import require_candidate_content
+
+            require_candidate_content(source, head)
         self._runner_commit = runner_head
         with tempfile.TemporaryDirectory(prefix="benchmark-source-") as directory:
             archive = Path(directory) / "source.tar"
@@ -247,6 +251,27 @@ async def _stage_source(self, environment: BaseEnvironment, source: Path) -> str
                 ), timeout_sec=180)
         return head
 
+    async def _install_research_entrypoint(self, environment, name: str, module: str,
+                                           help_flag: str, failure: str) -> None:
+        # Keep the task cwd without importing executable code or dependencies
+        # from the task workspace or user site.
+        entrypoint = f"{_SRC}/benchmark/runtime/{module}"
+        bootstrap = (
+            "import runpy\nimport sys\n"
+            f"sys.path.insert(0, {_SRC!r})\n"
+            f"runpy.run_path({entrypoint!r}, run_name='__main__')\n"
+        )
+        launcher = shlex.join([f"{_PYTHON}/bin/python3", "-I", "-c", bootstrap])
+        wrapper = f"#!/bin/sh\nexec {launcher} \"$@\"\n"
+        executable = shlex.quote(f"{_PROFILE}/bin/{name}")
+        installed = await self.exec_as_agent(
+            environment, env=self._profile_env(), timeout_sec=30,
+            command=f"test ! -L {executable} && printf %s {shlex.quote(wrapper)} > {executable} && "
+                    f"chmod 0755 {executable} && {executable} --help",
+        )
+        if installed.return_code or help_flag not in (installed.stdout or ""):
+            raise RuntimeError(failure)
+
     def _profile_env(self) -> dict[str, str]:
         return {
             "HOME": _PROFILE_HOME,
@@ -370,22 +395,10 @@ async def install(self, environment: BaseEnvironment) -> None:
             if doctor_payload.get("ok") is not True:
                 raise RuntimeError(f"LoopX doctor failed: {doctor_payload}")
             if self.execution_observation is not None:
-                entrypoint = f"{_SRC}/benchmark/runtime/research_probe.py"
-                # Keep the task cwd for candidate paths without importing the
-                # probe or its dependencies from that cwd or user site.
-                bootstrap = (
-                    "import runpy\nimport sys\n"
-                    f"sys.path.insert(0, {_SRC!r})\n"
-                    f"runpy.run_path({entrypoint!r}, run_name='__main__')\n"
+                await self._install_research_entrypoint(
+                    environment, "loopx-research-probe", "research_probe.py", "--candidate-file",
+                    "Local execution probe is unavailable to the benchmark worker",
                 )
-                launcher = shlex.join([f"{_PYTHON}/bin/python3", "-I", "-c", bootstrap])
-                wrapper = f"#!/bin/sh\nexec {launcher} \"$@\"\n"
-                probe = shlex.quote(f"{_PROFILE}/bin/loopx-research-probe")
-                installed = await self.exec_as_agent(environment, env=self._profile_env(), timeout_sec=30,
-                    command=f"test ! -L {probe} && printf %s {shlex.quote(wrapper)} > {probe} && "
-                            f"chmod 0755 {probe} && {probe} --help")
-                if installed.return_code or "--candidate-file" not in (installed.stdout or ""):
-                    raise RuntimeError("Local execution probe is unavailable to the benchmark worker")
 
         receipt = {
             "loopx_commit": actual_commit,

benchmark/runtime/research_probe.py

@@ -60,19 +60,16 @@ def execution_observation_guidance() -> str:
 
 
 def candidate_snapshot(paths: list[Path], workspace: Path) -> tuple[str, list[dict[str, str]]]:
-    files = []
-    for path in paths:
-        resolved = path.resolve(strict=True)
-        if not resolved.is_relative_to(workspace) or not resolved.is_file():
-            raise ValueError("candidate files must be existing files in the workspace")
-        with resolved.open("rb") as stream:
-            digest = hashlib.file_digest(stream, "sha256").hexdigest()
-        files.append({"path": resolved.relative_to(workspace).as_posix(), "sha256": digest})
-    if not 1 <= len(files) <= 32 or len({row["path"] for row in files}) != len(files):
-        raise ValueError("select one to 32 distinct candidate files")
-    files.sort(key=lambda row: row["path"])
-    encoded = json.dumps(files, sort_keys=True, separators=(",", ":"))
-    return hashlib.sha256(encoded.encode()).hexdigest(), files
+    # Lazy import keeps feature-off runners compatible with older product pins.
+    from loopx.capabilities.explore.candidate_content import candidate_snapshot as capture
+
+    return capture(paths, workspace)
+
+
+def candidate_manifest_sha256(files: list[dict[str, str]]) -> str:
+    from loopx.capabilities.explore.candidate_content import candidate_manifest_sha256 as digest
+
+    return digest(files)
 
 
 def _group_members(group_id: int) -> dict[int, bytes]:

benchmark/runtime/sforge.py

@@ -377,6 +377,11 @@ def _install_worker(self, backend, handle, log_dir, logger):
                     "defer_codex_home": True} if self.warm_start is not None else {}),
             )
             asyncio.run(self.runtime.install(self.environment))
+            if self.feedback == "native" and self.execution_observation is not None:
+                asyncio.run(self.runtime._install_research_entrypoint(
+                    self.environment, "loopx-bind-native-candidate", "sforge_candidate_binding.py",
+                    "--generated-file", "Native candidate verification is unavailable to the benchmark worker",
+                ))
             if self.warm_start is not None:
                 route_settings = {}
                 if runtime_base_url: