fix(benchmark): restore worker access to selected research data

benchmark/runtime/sforge.py

@@ -281,11 +281,13 @@ def _install_worker(self, backend, handle, log_dir, logger):
             )
             asyncio.run(self.runtime.install(self.environment))
             if self.warm_start is not None:
-                asyncio.run(self.warm_start.stage(self.environment, self.cwd))
+                research_paths = asyncio.run(self.warm_start.stage(self.environment, self.cwd))
                 self._warm_command("workspace", user="root")
-                result = backend.exec_run(handle, ["chown", "-R", "-P", "agent:agent", self.cwd], user="root")
+                result = backend.exec_run(handle,
+                    ["chown", "-R", "-P", "agent:agent", self.cwd, *research_paths], user="root")
                 if result.exit_code:
-                    raise RuntimeError("Restored workspace is not writable by its receiver")
+                    raise RuntimeError("Restored research state is not writable by its receiver"
+                        if research_paths else "Restored workspace is not writable by its receiver")
             if self.checkpoint_request is not None:
                 with tempfile.TemporaryDirectory(prefix="benchmark-checkpoint-") as directory:
                     declaration = Path(directory) / "request.json"

benchmark/runtime/sforge_warm_start.py

@@ -87,8 +87,9 @@ def verify(self, workspace: str) -> dict:
             raise ValueError("Original warm start deadline expired during checkpoint verification")
         return metadata
 
-    async def stage(self, environment, workspace: str):
-        self.verify(workspace)
+    async def stage(self, environment, workspace: str) -> list[str]:
+        metadata = self.verify(workspace)
+        research = _research_components(self.root, metadata, Path(workspace), self.research_component_ids)
         # Install tools first without creating the native home. Stage inert
         # participants outside the workspace so restoring its bytes cannot
         # replace the active controller, registry or installed executable.
@@ -115,6 +116,10 @@ async def stage(self, environment, workspace: str):
             f"chmod 0400 {config} {_CONTROL}/warm-start.json", user="root")
         if result.return_code:
             raise RuntimeError("Could not stage the verified checkpoint participants")
+        # The root-only materializer preserves sealed modes. Its receiver must
+        # transfer these selected data copies to the worker before preparation;
+        # making private directories world-readable would change the artifact.
+        return [str(target) for _, _, target in research]
 
 
 # Provider-owned data locations; selecting a component never grants permission

benchmark/tests/test_sforge_runtime.py

@@ -180,6 +180,53 @@ async def installed(self, environment):
         SForgeWorker(config, profile="official", cwd="/task", checkpoint_request_path="/control/request.json")
 
 
+@pytest.mark.parametrize("research_paths", [(), ("/tmp/private-data.json",), ("/tmp/private-research",)])
+@pytest.mark.parametrize("refuse_transfer", [False, True])
+def test_warm_receiver_owns_only_selected_research_before_preparation(
+        tmp_path, monkeypatch, research_paths, refuse_transfer):
+    pytest.importorskip("sforge")
+    pytest.importorskip("harbor")
+    from types import SimpleNamespace
+    from sforge.harness.config import SForgeConfig
+    from benchmark.runtime.sforge import BenchmarkCodex, SForgeWorker
+
+    monkeypatch.setenv("CODEX_AUTH_JSON_PATH", "/synthetic-credential")
+    async def installed(self, environment):
+        pass
+    monkeypatch.setattr(BenchmarkCodex, "install", installed)
+    async def stage(environment, workspace):
+        return list(research_paths)  # Verified selection is exercised by WarmStart staging tests.
+    start = SimpleNamespace(goal_id="receiver", agent_id="worker", stage=stage,
+                            verify=lambda workspace: None)
+    owned = set()
+    commands = []
+    def execute(handle, command, **kwargs):
+        commands.append(command)
+        if command[:4] == ["chown", "-R", "-P", "agent:agent"]:
+            assert kwargs["user"] == "root"
+            if refuse_transfer:
+                return SimpleNamespace(exit_code=1)
+            owned.update(command[4:])
+        return SimpleNamespace(exit_code=0, output="")
+    worker = SForgeWorker(SForgeConfig(agent_model="fixture", agent_effort="xhigh"),
+        profile="heartbeat-explore", cwd="/task", warm_start=start)
+    restored = []
+    monkeypatch.setattr(worker, "_warm_command", lambda action, **kwargs: restored.append(action))
+    backend = SimpleNamespace(exec_run=execute, copy_to_container=lambda *args: None)
+    if refuse_transfer:
+        with pytest.raises(RuntimeError, match="receiver") as failure:
+            worker.install_stop_hook(backend, None, tmp_path, None)
+        if not research_paths:
+            assert str(failure.value) == "Restored workspace is not writable by its receiver"
+        assert not (tmp_path / "worker-profile.json").exists()
+    else:
+        worker.install_stop_hook(backend, None, tmp_path, None)
+        assert owned == {"/task", *research_paths}
+        assert "/tmp/unselected-research" not in owned
+        assert not any(command[0] == "chmod" for command in commands)
+    assert restored == ["workspace"]
+
+
 @pytest.mark.parametrize("explicit,trigger,expected", [
     (False, None, None),
     (False, "fresh_replan", None),

benchmark/tests/test_sforge_warm_start.py

@@ -188,7 +188,7 @@ async def upload_file(self, source, target):
         async def exec(self, command, **kwargs):
             commands.append(command)
             return SimpleNamespace(return_code=0)
-    asyncio.run(start.stage(Environment(), str(workspace)))
+    assert asyncio.run(start.stage(Environment(), str(workspace))) == []
     assert "chmod -R" not in commands[0]
     effect_runtime_result("runtime.checkpoint_artifacts", {"action": "verify", "root": str(control / "restored-point"),
         "expected_sha256": start.sha256})
@@ -300,6 +300,7 @@ def research_point(point, tmp_path, monkeypatch, request):
         research.chmod(0o600)
     else:
         research.mkdir()
+        research.chmod(0o700)
         (research / "candidate.py").write_text("def predict(x): return x * 7\n")
         (research / "observations.json").write_text('[2, 3, 5]')
         (research / "candidate.py").chmod(0o600)
@@ -409,8 +410,9 @@ def corrupt(source, destination, *args, **kwargs):
     assert workspace.with_name("workspace.before-checkpoint").is_dir()
 
 
+@pytest.mark.parametrize("research_point", ["directory", "file"], indirect=True)
 def test_research_selection_survives_staging(research_point, monkeypatch, tmp_path):
-    start, workspace, _ = research_point
+    start, workspace, research = research_point
     control = tmp_path / "control"
     control.mkdir()
     monkeypatch.setattr(warm, "_CONTROL", str(control))
@@ -421,7 +423,7 @@ async def upload_file(self, source, target):
             shutil.copyfile(source, target)
         async def exec(self, command, **kwargs):
             return SimpleNamespace(return_code=0)
-    asyncio.run(start.stage(Environment(), str(workspace)))
+    assert asyncio.run(start.stage(Environment(), str(workspace))) == [str(research)]
     declaration = json.loads((control / "warm-start.json").read_bytes())
     assert declaration["research_component_ids"] == ["experiments"]