fix(benchmark): retain research outside native submission trees

benchmark/edgebench/README.md

@@ -671,6 +671,28 @@ and credentials belong outside the public repository.
 
 ## Research checkpoint storage
 
+New native non-game runs give every worker profile the same research directory,
+`/tmp/loopx-benchmark-research`, outside the task workspace and submitted paths.
+Keep experiment notes, local probes and candidate history there; required build
+and runtime inputs still belong in the task's declared submission paths. The
+runner does not filter submissions, extend evaluator feedback or alter grading.
+Overlapping submission/capture roots and existing unrelated data at this reserved
+path reject setup before solver execution.
+
+Terminal collection retains this directory under `collected/research`. With an
+explicit checkpoint request, the runner adds a separate `benchmark_research`
+component to the staged declaration, without modifying the caller's file. Warm
+continuation restores that exact component at its original path through the
+ordinary verified research-file transport. Other external research components
+still require explicit selection. Old checkpoints remain valid and preserve
+their workspace files; they start with an empty external research directory,
+without automatic relocation or deletion of historical artifacts. A warm receiver
+appends a bounded storage-policy update to its existing `AGENTS.md`, preserving
+the original task text and prior feedback policy; it does not rewrite the sealed
+checkpoint or Goal state. Terminal
+collection alone does not make a run resumable. Game and restricted-feedback
+runs retain their existing storage behavior.
+
 Full controller capture requires a filesystem that preserves case-sensitive and
 Unicode-normalization-sensitive filenames. A Linux controller using a shared
 host volume may not have those properties. New controller-enabled launches probe

benchmark/edgebench/prompts.py

@@ -6,8 +6,6 @@
 
 from __future__ import annotations
 
-from pathlib import PurePosixPath
-
 
 def native_research_guidance() -> str:
     """Keep the task's optimization objective explicit."""
@@ -19,38 +17,31 @@ def native_research_guidance() -> str:
     )
 
 
-def checkpoint_research_instructions(request: dict, workspace: str) -> str:
-    """Describe declared coverage without inferring or expanding capture authority.
+def research_artifact_instructions(*, checkpoint: bool) -> str:
+    """Separate reproducibility data from candidate inputs for every native arm."""
+    from benchmark.runtime.research_artifacts import RESEARCH_DIRECTORY
 
-    Directory membership remains the artifact owner's responsibility at capture.
-    This provider guidance keeps newly generated scientific outputs in that scope;
-    it neither proves dependency closure nor changes the evaluator's file set.
-    """
-    root = PurePosixPath(workspace)
-    covered = any(
-        isinstance(entry, dict) and isinstance(entry.get("source_path"), str)
-        and root.is_relative_to(PurePosixPath(entry["source_path"]))
-        for entry in request["components"]
-    )
-    coverage = (
-        "The checkpoint declaration includes this task workspace. New files saved "
-        "inside that directory are included when it is captured. "
-        if covered else
-        "The checkpoint declaration does not cover the whole task workspace. "
-        "Do not assume new research files are captured; report the missing directory "
-        "coverage before relying on continuation. "
+    capture = (
+        "Checkpoint capture includes this directory as a separate component; "
+        "verified continuation restores its files at the same path. "
+        if checkpoint else
+        "Terminal artifact collection retains this directory for analysis; "
+        "that alone does not provide a resumable checkpoint. "
     )
     return (
-        "### Durable Research Artifacts\n\n"
-        "Checkpoint capture is enabled. " + coverage +
-        "Keep reusable research artifacts in a dedicated subdirectory of the task "
-        "workspace: candidate snapshots, fitted parameters, local test inputs, "
+        "### Research Artifacts\n\n"
+        f"Use `{RESEARCH_DIRECTORY}` for reusable research artifacts that are not "
+        "candidate inputs: candidate snapshots, fitted parameters, local test inputs, "
         "experiment reports and negative results needed to reproduce or reconsider "
-        "a decision. Use workspace-relative references where possible. Reserve "
-        "temporary directories for disposable scratch; a report that only names an "
+        "a decision. This directory is outside the task workspace and submitted paths. "
+        "Keep required candidate code and runtime/build data in the task's declared "
+        "submitted paths; the judge cannot use this research directory as an extra input. "
+        "Do not link it into the submission tree. " + capture +
+        "Use explicit references to retained artifacts. A report that only names an "
         "external file does not preserve its bytes. Keep existing external dependencies "
-        "explicit until they have been copied and their references updated and checked; "
-        "do not delete or silently substitute historical evidence.\n\n"
+        "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"
         "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

@@ -39,8 +39,8 @@
 from benchmark.runtime.sforge_backend import RecordingDockerBackend
 from benchmark.runtime.source import archive_source, source_pins
 from benchmark.edgebench.prompts import (
-    blind_task_prompt, best_only_task_prompt, native_task_prompt, native_task_instructions,
-    checkpoint_research_instructions,
+    blind_task_prompt, best_only_task_prompt, native_task_instructions,
+    research_artifact_instructions,
 )
 from benchmark.edgebench.online_sampling import OnlineSampler
 from benchmark.edgebench.feedback import BestOnlyFeedback, FEEDBACK_MODES, validate_best_only
@@ -277,6 +277,13 @@ def main(argv=None):
                 archive_source(runner, pins[1], Path(directory) / "runner.tar", ("benchmark/runtime",))
     task_file = args.tasks_dir / f"{args.task}.json"
     task = make_task_spec(task_file, load_benchmark(args.tasks_dir))
+    research_sidecar = args.feedback == "native" and not task.game_mode
+    if research_sidecar:
+        from benchmark.runtime.research_artifacts import validate_research_scope, with_research_capture
+        validate_research_scope(task.cwd, task.submit_paths)
+        if checkpoint_request is not None:
+            declaration = with_research_capture(declaration)
+            checkpoint_request = json.dumps(declaration).encode("utf-8")
     blind_endpoint = None
     feedback_prompt = None
     workspace_instructions = None
@@ -300,14 +307,11 @@ def main(argv=None):
         native_policy = dict(submission_cooldown=args.submission_cooldown,
             eval_interval=args.eval_interval, internet=task.internet,
             selection=task.judge.selection, score_direction=task.judge.score_direction)
-        feedback_prompt = native_task_prompt(task.work.agent_query, task.submit_paths, **native_policy)
         workspace_instructions = native_task_instructions(task.submit_paths, **native_policy)
-        if checkpoint_request is not None:
-            research = checkpoint_research_instructions(declaration, task.cwd)
-            workspace_instructions += research
-            # Keep the exact policy prefix shared with AGENTS.md; the original
-            # task remains the separately owned Goal objective.
-            feedback_prompt = workspace_instructions + f"---\n\n{task.work.agent_query}\n"
+        workspace_instructions += research_artifact_instructions(checkpoint=checkpoint_request 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"
         task_objective = task.work.agent_query
     config = replace(
         load_config(),
@@ -344,7 +348,7 @@ def main(argv=None):
             args.restore_goal, args.restore_agent, args.restore_source_todo,
             int(time.time()) + args.timeout, args.replan_after_turns or 6,
             tuple(args.restore_research_component), args.restore_fresh_replan,
-            destination_config=args.restore_model_config)
+            destination_config=args.restore_model_config, research_sidecar=research_sidecar)
         metadata = candidate.verify(task.cwd)
         if args.restore_model_config is not None:
             from loopx.hosts.codex_checkpoint import checkpoint_destination_config
@@ -358,7 +362,7 @@ def main(argv=None):
         clock = attempt / "restored-phase-deadline"
         warm_start = candidate
     agent = SForgeWorker(config, profile=args.worker, cwd=task.cwd,
-                         warm_start=warm_start,
+                         warm_start=warm_start, research_sidecar=research_sidecar,
                          timeout_seconds=args.timeout, feedback_prompt=feedback_prompt, feedback=args.feedback,
                          workspace_instructions=workspace_instructions,
                          task_objective=task_objective,
@@ -403,6 +407,7 @@ def main(argv=None):
                                      oauth_proxy=bool(args.api_proxy_url or api_base_url),
                                      blind_api_endpoint=blind_endpoint, feedback=feedback,
                                      docker_host_bridge=args.docker_host_bridge,
+                                     research_sidecar=research_sidecar,
                                      **({"checkpoint": checkpoint} if checkpoint is not None else {}))
     for image in (task.work_image_key, task.judge_image_key):
         if not backend.image_exists(image):

benchmark/runtime/research_artifacts.py

@@ -0,0 +1,66 @@
+"""Provider-owned research data outside the native submission tree.
+
+This composes the existing checkpoint component transport. It neither filters
+submitted files nor changes the task's evaluator or feedback permissions.
+"""
+from __future__ import annotations
+
+from copy import deepcopy
+from pathlib import PurePosixPath
+
+
+RESEARCH_DIRECTORY = "/tmp/loopx-benchmark-research"
+RESEARCH_COMPONENT = "benchmark_research"
+
+
+def validate_research_scope(workspace: str, submit_paths: list[str]) -> None:
+    """Refuse an overlapping task instead of silently changing submission scope."""
+    root = PurePosixPath(workspace)
+    research = PurePosixPath(RESEARCH_DIRECTORY)
+    roots = [root, *(PurePosixPath(path) if PurePosixPath(path).is_absolute()
+                     else root / path for path in submit_paths)]
+    for path in roots:
+        if not path.is_absolute() or ".." in path.parts:
+            raise ValueError("Research storage requires normalized absolute task and submission paths")
+        if research.is_relative_to(path) or path.is_relative_to(research):
+            raise ValueError("Research storage overlaps the task or its submitted paths")
+
+
+def with_research_capture(request: dict) -> dict:
+    """Add this runner's data to a copy of its explicit capture declaration."""
+    result = deepcopy(request)
+    expected = {"id": RESEARCH_COMPONENT, "source_path": RESEARCH_DIRECTORY}
+    present = False
+    for component in result["components"]:
+        if component.get("id") == RESEARCH_COMPONENT:
+            if component != expected or present:
+                raise ValueError("Reserved research component conflicts with the checkpoint declaration")
+            present = True
+        else:
+            source = PurePosixPath(component["source_path"])
+            research = PurePosixPath(RESEARCH_DIRECTORY)
+            if source.is_relative_to(research) or research.is_relative_to(source):
+                raise ValueError("Checkpoint component overlaps the reserved research directory")
+    if not present:
+        result["components"].append(expected)
+    return result
+
+
+def selected_research_components(metadata: dict, selected, *, managed: bool):
+    """Only the runner's exact sealed component joins explicit receiver choices.
+
+    Older checkpoints do not have this component. Their original workspace and
+    explicit research selections remain intact; no historical files are moved.
+    """
+    if not managed:
+        return selected
+    if (not isinstance(selected, (tuple, list)) or
+            any(not isinstance(key, str) for key in selected) or len(set(selected)) != len(selected)):
+        raise ValueError("Research component selection must contain distinct captured IDs")
+    matches = [entry for entry in metadata["worker_file_snapshot"]["components"]
+               if entry["id"] == RESEARCH_COMPONENT]
+    if not matches:
+        return selected
+    if len(matches) != 1 or matches[0] != {"id": RESEARCH_COMPONENT, "source_path": RESEARCH_DIRECTORY}:
+        raise ValueError("Captured runner research component has a conflicting identity")
+    return tuple(selected) if RESEARCH_COMPONENT in selected else (*selected, RESEARCH_COMPONENT)

benchmark/runtime/sforge.py

@@ -101,7 +101,7 @@ def __init__(self, config, *, profile: str, cwd: str,
                  planning_horizon_goal: bool | None = None,
                  checkpoint_request_path: str | None = None,
                  checkpoint_request: bytes | None = None,
-                 warm_start=None):
+                 warm_start=None, research_sidecar: bool = False):
         super().__init__(config)
         if profile not in PROFILES:
             raise ValueError("Unknown benchmark worker profile")
@@ -184,8 +184,13 @@ def __init__(self, config, *, profile: str, cwd: str,
                 raise ValueError("checkpoint_request_path requires a heartbeat profile")
             from .checkpoint import normalize_checkpoint_request_path
             self.checkpoint_request_path = normalize_checkpoint_request_path(checkpoint_request_path)
+        if type(research_sidecar) is not bool:
+            raise ValueError("research_sidecar requires an explicit boolean")
+        self.research_sidecar = research_sidecar
         self.warm_start = warm_start
         if warm_start is not None:
+            if research_sidecar and not warm_start.research_sidecar:
+                raise ValueError("Managed research storage requires matching warm receiver selection")
             if not profile.startswith("heartbeat-") or feedback != "native" or self.task_entry != "loopx-planned":
                 raise ValueError("Warm checkpoint continuation requires a planned native heartbeat worker")
             warm_start.verify(cwd)
@@ -242,9 +247,26 @@ def install_stop_hook(self, backend, handle, log_dir, logger):
             # Solver execution must never inherit this setup-only proxy.
             self.environment.setup_env = {}
 
+    def _install_research_directory(self, backend, handle, *, restored: bool):
+        from .research_artifacts import RESEARCH_DIRECTORY
+        directory = shlex.quote(RESEARCH_DIRECTORY)
+        # Fresh images cannot contribute unrelated data at this reserved path.
+        # Warm restore has already verified bytes and modes; do not chmod them.
+        prepare = (f"test -d {directory} && test ! -L {directory}" if restored else
+                   f"mkdir -m 0700 {directory}")
+        result = backend.exec_run(handle, ["/bin/sh", "-c",
+            f'test "$(readlink -f /tmp)" = /tmp && '
+            f'workspace=$(readlink -f -- {shlex.quote(self.cwd)}) && test -n "$workspace" && '
+            f'case {directory}/ in "$workspace/"*) exit 1;; esac && '
+            f'case "$workspace/" in {directory}/*) exit 1;; esac && '
+            f'{prepare} && chown agent:agent {directory}'], user="root")
+        if result.exit_code:
+            raise RuntimeError("Research directory is unavailable or conflicts with existing task data")
+
     def _install_worker(self, backend, handle, log_dir, logger):
         self.backend, self.handle = backend, handle
         self.log_dir = log_dir
+        research_paths = []
         effort = self._config.agent_effort
         common = dict(logs_dir=log_dir / "worker", model_name=self._config.agent_model,
                       reasoning_effort=effort)
@@ -362,6 +384,9 @@ def _install_worker(self, backend, handle, log_dir, logger):
                     f"chmod 0600 {shlex.quote(self.checkpoint_request_path)}"], user="root")
                 if result.exit_code:
                     raise RuntimeError("Could not protect the explicit checkpoint declaration")
+        if self.research_sidecar:
+            from .research_artifacts import RESEARCH_DIRECTORY
+            self._install_research_directory(backend, handle, restored=RESEARCH_DIRECTORY in research_paths)
         # Prompt replacement describes the policy; the feedback mode owns access.
         if self.feedback != "native":
             result = backend.exec_run(handle, ["rm", "-f", "/usr/local/bin/sforge-submit"], user="root")
@@ -397,6 +422,7 @@ def _install_worker(self, backend, handle, log_dir, logger):
             "task_entry": self.task_entry,
             "reasoning_effort": effort, "timeout_seconds": self.timeout_seconds,
             "loopx_usage_ping_enabled": False,
+            **({"research_directory": RESEARCH_DIRECTORY} if self.research_sidecar else {}),
             **({"checkpoint_capture_enabled": True}
                if self.checkpoint_request_path is not None else {}),
             **({"planning_timeout_seconds": None}
@@ -416,8 +442,11 @@ def _install_worker(self, backend, handle, log_dir, logger):
         }, indent=2))
 
     def format_run_cmd(self, prompt_path, *, model=None, cwd="", internet=True, resume=False):
-        if self.workspace_instructions is not None and not self.instructions_installed and self.warm_start is None:
-            self._install_workspace_instructions()
+        if self.workspace_instructions is not None and not self.instructions_installed:
+            if self.warm_start is None:
+                self._install_workspace_instructions()
+            elif self.research_sidecar:
+                self._install_restored_research_instructions()
         if self.feedback_prompt is not None and not self.prompt_installed:
             with tempfile.TemporaryDirectory(prefix="benchmark-prompt-") as directory:
                 prompt = Path(directory) / "task.md"
@@ -534,9 +563,21 @@ def _warm_command(self, action, *, user="agent"):
 
     def _install_workspace_instructions(self):
         """Preserve task instructions and verify the policy before model entry."""
+        self._append_workspace_policy("<!-- EdgeBench native feedback policy -->", self.workspace_instructions)
+
+    def _install_restored_research_instructions(self):
+        from benchmark.edgebench.prompts import research_artifact_instructions
+        # Preserve the sealed task instructions and earlier feedback policy.
+        # Only the runner-owned storage location changes in this continuation;
+        # the original checkpoint remains untouched and independently verifiable.
+        self._append_workspace_policy("<!-- EdgeBench restored research storage -->",
+            "For this continuation, the following research storage instructions "
+            "replace earlier workspace-storage guidance. Existing implementation and "
+            "research artifacts are retained.\n\n" + research_artifact_instructions(checkpoint=True))
+
+    def _append_workspace_policy(self, marker: str, policy: str):
         path = shlex.quote(str(PurePosixPath(self.cwd) / "AGENTS.md"))
-        marker = "<!-- EdgeBench native feedback policy -->"
-        block = f"\n\n{marker}\n\n{self.workspace_instructions}"
+        block = f"\n\n{marker}\n\n{policy}"
         check = f"test ! -L {path} && {{ test ! -e {path} || test -f {path}; }}"
         read = asyncio.run(self.environment.exec(f"{check} && {{ test ! -f {path} || cat {path}; }}"))
         if read.return_code: