fix(benchmark): guide checkpoint research into declared workspace

benchmark/edgebench/prompts.py

@@ -6,6 +6,8 @@
 
 from __future__ import annotations
 
+from pathlib import PurePosixPath
+
 
 def native_research_guidance() -> str:
     """Encourage continued research without replacing the task-owned objective."""
@@ -16,6 +18,44 @@ def native_research_guidance() -> str:
     )
 
 
+def checkpoint_research_instructions(request: dict, workspace: str) -> str:
+    """Describe declared coverage without inferring or expanding capture authority.
+
+    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. "
+    )
+    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, "
+        "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 "
+        "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"
+        "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"
+    )
+
+
 def native_task_instructions(submit_paths: list[str], *,
                              submission_cooldown: int, eval_interval: int, internet: bool,
                              selection: str, score_direction: str) -> str:

benchmark/edgebench/run.py

@@ -29,6 +29,7 @@
 from benchmark.runtime.source import source_pins
 from benchmark.edgebench.prompts import (
     blind_task_prompt, best_only_task_prompt, native_task_prompt, native_task_instructions,
+    checkpoint_research_instructions,
 )
 from benchmark.edgebench.online_sampling import OnlineSampler
 from benchmark.edgebench.feedback import BestOnlyFeedback, FEEDBACK_MODES, validate_best_only
@@ -236,6 +237,12 @@ def main(argv=None):
             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"
         task_objective = task.work.agent_query
     config = SForgeConfig(
         agent_api_base_url=api_base_url, agent_api_key=api_key,

benchmark/tests/test_checkpoint_worker.py

@@ -607,3 +607,38 @@ def broken(*args):
     supervisor._run(container)
     assert len(reads) == 3 and isinstance(supervisor.failure, tarfile.ReadError)
     assert json.loads((tmp_path / "supervisor-failure.json").read_text()) == {"error_kind": "ReadError"}
+
+
+def test_declared_directory_includes_research_created_after_declaration(controllers, writer, tmp_path):
+    from loopx.control_plane.effect_runtime import effect_runtime_result
+    source, _, first, _, files, grades = controllers
+    container, request = writer
+    token = register(source, first)
+    # writer already declared /data. New artifacts need no new component IDs.
+    payload = json.loads(request.read_text())
+    assert payload["components"] == [{"id": "native_capsule", "source_path": "/data"}]
+    assert container.exec_run(["python", "-c", "from pathlib import Path; "
+        "p=Path('/data/research/new-method'); p.mkdir(parents=True); "
+        "(p/'candidate.py').write_text('candidate'); "
+        "(p/'negative.json').write_text('{\"accepted\": false}'); "
+        "Path('/outside-report.json').write_text('not declared')"]).exit_code == 0
+    unit = tmp_path / "directory-point"
+    result = capture_worker(files[0], token=token, capture_id="f" * 32,
+                           destination=unit, request_file=request)
+    assert result["integrity_verified"]
+    assert (unit / "components/native_capsule/research/new-method/candidate.py").read_text() == "candidate"
+    assert json.loads((unit / "components/native_capsule/research/new-method/negative.json").read_text()) == {"accepted": False}
+    assert not list(unit.rglob("outside-report.json"))
+    effect_runtime_result("runtime.checkpoint_artifacts", {"action": "verify", "root": str(unit),
+        "expected_sha256": result["sha256"]})
+    restored = tmp_path / "restored"
+    receipt = effect_runtime_result("runtime.checkpoint_artifacts", {
+        "action": "materialize", "root": str(unit),
+        "expected_sha256": result["sha256"], "destination": str(restored),
+    })
+    assert receipt["materialized"] and not receipt["activation_performed"]
+    research = restored / "components/native_capsule/research/new-method"
+    assert (research / "candidate.py").read_text() == "candidate"
+    assert json.loads((research / "negative.json").read_text()) == {"accepted": False}
+    assert not list(restored.rglob("outside-report.json"))
+    assert request.read_text() == json.dumps(payload) and not grades

benchmark/tests/test_edgebench_feedback_docker.py

@@ -819,3 +819,51 @@ def do_POST(self):
     finally:
         server.shutdown()
         thread.join(timeout=5)
+
+
+def test_checkpoint_storage_policy_is_installed_in_real_worker(tmp_path, agent_image, monkeypatch):
+    from sforge.harness.config import SForgeConfig
+    from benchmark.runtime.sforge import SForgeWorker, SForgeEnvironment
+    from benchmark.runtime.sforge_backend import RecordingDockerBackend
+    from benchmark.edgebench.prompts import native_task_instructions, checkpoint_research_instructions
+    monkeypatch.setenv("CODEX_AUTH_JSON_PATH", "/synthetic-unused-credential")
+    request = {"destination": "/logs/agent/checkpoints", "point_metadata": {},
+               "components": [{"id": "workspace", "source_path": "/task"}]}
+    policy = native_task_instructions(["solver.py"], submission_cooldown=3600,
+        eval_interval=0, internet=False, selection="score_first", score_direction="maximize")
+    policy += checkpoint_research_instructions(request, "/task")
+    worker = SForgeWorker(SForgeConfig(agent_model="fixture", agent_effort="xhigh"),
+        profile="heartbeat-explore", cwd="/task", checkpoint_request=json.dumps(request).encode(),
+        workspace_instructions=policy, feedback_prompt=policy + "---\n\nOriginal acceptance.\n")
+    backend = RecordingDockerBackend(log_dir=tmp_path / "collected",
+        logger=logging.getLogger("storage-policy-smoke"), oauth_proxy=True)
+    handle = backend.create_container(agent_image.id, "storage-policy-" + uuid.uuid4().hex[:10],
+                                      cpu_limit=1, mem_limit="256m")
+    backend.start_container(handle)
+    try:
+        result = backend.exec_run(handle, ["sh", "-c",
+            "mkdir /task && printf 'Original task instructions.' > /task/AGENTS.md && chown -R agent:agent /task"], user="root")
+        assert result.exit_code == 0
+        worker.environment = SForgeEnvironment(backend, handle)
+        worker.log_dir = tmp_path
+        # Production prompt/AGENTS installation precedes the missing runtime
+        # guard. No bootstrap, native session or model is started by this test.
+        with pytest.raises(RuntimeError, match="installation hook"):
+            worker.format_run_cmd("/task/prompt.md")
+        agents = backend.exec_run(handle, ["cat", "/task/AGENTS.md"], user="agent")
+        prompt = backend.exec_run(handle, ["cat", "/task/prompt.md"], user="agent")
+        assert agents.exit_code == prompt.exit_code == 0
+        assert agents.output.startswith("Original task instructions.")
+        assert agents.output.endswith(policy)
+        assert prompt.output == worker.feedback_prompt
+        assert (tmp_path / "workspace-AGENTS.md").read_text() == agents.output
+        # Reapplying matching policy is idempotent; conflicting installed text
+        # fails before replacing the original task instructions.
+        worker._install_workspace_instructions()
+        worker.workspace_instructions += "Changed policy"
+        with pytest.raises(RuntimeError, match="conflicting or partial"):
+            worker._install_workspace_instructions()
+        assert backend.exec_run(handle, ["cat", "/task/AGENTS.md"], user="agent").output == agents.output
+    finally:
+        backend.cleanup_container(handle)
+        backend.client.close()

benchmark/tests/test_edgebench_prompts.py

@@ -387,3 +387,20 @@ def test_workspace_policy_cannot_drift_from_feedback_mode_or_startup(
         SForgeWorker(SForgeConfig(agent_model="fixture", agent_effort="xhigh"),
             profile="official", cwd="/task", feedback=feedback,
             feedback_prompt=prompt, workspace_instructions=instructions)
+
+
+@pytest.mark.parametrize("source,covered", [
+    ("/task", True), ("/home/workspace", True),
+    ("/home/workspace/task/solver.py", False), ("/home/workspace/task-other", False),
+])
+def test_checkpoint_guidance_distinguishes_directory_from_file_coverage(source, covered):
+    from benchmark.edgebench.prompts import checkpoint_research_instructions
+    workspace = "/task" if source == "/task" else "/home/workspace/task"
+    text = checkpoint_research_instructions(
+        {"components": [{"id": "code", "source_path": source}]}, workspace)
+    assert ("includes this task workspace" in text) is covered
+    assert ("does not cover the whole task workspace" in text) is not covered
+    assert "negative results" in text and "workspace-relative references" in text
+    assert "external file does not preserve its bytes" in text
+    assert "does not change submitted files" in text
+    assert source not in text  # Do not echo arbitrary private capture roots.

benchmark/tests/test_sforge_runtime.py

@@ -1030,3 +1030,47 @@ def stop_before_solver(**kwargs):
                   '--effort', 'xhigh', '--judge-url', 'http://127.0.0.1:9999'])
     receipt = (tmp_path / 'runs/route/fixture/runtime-receipt.json').read_text()
     assert 'synthetic-provider-secret' not in receipt
+
+
+@pytest.mark.parametrize("capture", [False, True])
+def test_native_checkpoint_storage_guidance_reaches_both_prompt_surfaces(tmp_path, monkeypatch, capture):
+    pytest.importorskip("sforge")
+    pytest.importorskip("harbor")
+    from types import SimpleNamespace
+    from benchmark.edgebench import run
+    from benchmark.edgebench.prompts import native_task_prompt, native_task_instructions
+
+    monkeypatch.setenv("LOOPX_SRC_DIR", str(tmp_path))
+    monkeypatch.setenv("LOOPX_EXPECTED_COMMIT", "fixture")
+    monkeypatch.setenv("CODEX_AUTH_JSON_PATH", "/synthetic-unused-credential")
+    (tmp_path / "fixture.json").write_text("{}")
+    monkeypatch.setattr(run, "source_pins", lambda *a: ("fixture", "fixture"))
+    monkeypatch.setattr(run, "load_benchmark", lambda *a: None)
+    monkeypatch.setattr(run, "make_task_spec", lambda *a: SimpleNamespace(
+        cwd="/task", work_image_key="work", judge_image_key="judge", internet=False,
+        game_mode=False, submit_paths=["solver.py"], work=SimpleNamespace(agent_query="Original acceptance."),
+        judge=SimpleNamespace(selection="score_first", score_direction="maximize")))
+    monkeypatch.setattr(run, "RecordingDockerBackend", lambda **k: SimpleNamespace(image_exists=lambda image: True))
+    def before_solver(**kwargs):
+        worker = kwargs["agent"]
+        assert worker.task_objective == "Original acceptance."
+        assert worker.feedback_prompt.startswith(worker.workspace_instructions)
+        assert worker.feedback_prompt.endswith("---\n\nOriginal acceptance.\n")
+        assert ("### Durable Research Artifacts" in worker.workspace_instructions) is capture
+        if not capture:
+            policy = dict(submission_cooldown=3600, eval_interval=0, internet=False,
+                          selection="score_first", score_direction="maximize")
+            assert worker.workspace_instructions == native_task_instructions(["solver.py"], **policy)
+            assert worker.feedback_prompt == native_task_prompt("Original acceptance.", ["solver.py"], **policy)
+        raise RuntimeError("validated before model execution")
+    monkeypatch.setattr(run, "run_agent", before_solver)
+    args = ["--task", "fixture", "--tasks-dir", str(tmp_path), "--log-dir", str(tmp_path),
+            "--run-id", "storage", "--worker", "heartbeat-explore", "--model", "fixture",
+            "--effort", "xhigh", "--judge-url", "http://127.0.0.1:9999", "--feedback", "native"]
+    if capture:
+        declaration = tmp_path / "request.json"
+        declaration.write_text(json.dumps({"destination": "/logs/agent/checkpoints",
+            "components": [{"id": "workspace", "source_path": "/task"}], "point_metadata": {}}))
+        args += ["--checkpoint-request", str(declaration)]
+    with pytest.raises(RuntimeError, match="validated before model execution"):
+        run.main(args)

docs/reference/research-checkpoint.md

@@ -55,6 +55,20 @@ explicitly materialize their selected research components; the
 [SForge warm transport](../../benchmark/README.md#explicit-warm-worker-composition-qualification-in-progress)
 provides that selection for its data paths.
 
+For new checkpoint-enabled native EdgeBench runs, the startup prompt and workspace
+`AGENTS.md` now tell the solver to keep reusable research artifacts in a task
+workspace subdirectory, including local inputs and negative results. The prompt
+states whether the supplied declaration covers that workspace. A declared
+directory includes files created later; an explicit file list does not grow
+when a conversation or report mentions another path. Capture-disabled prompts
+are unchanged. This is storage guidance, not an enforced completeness check.
+
+For new donors, declare the task workspace directory before launch. For existing
+runs, preserve their frozen instructions and source; use a newly qualified
+attempt when adopting this guidance. Previously external artifacts still need
+explicit capture and receiver validation. Do not rewrite an old point using
+later live files or claim that the new guidance repairs missing historical bytes.
+
 Verify the returned digest before preparing a receiver:
 
 ```sh