fix(benchmark): restore task workspace from captured parent directory

benchmark/README.md

@@ -959,8 +959,14 @@ All six inputs are required together. Keep the existing native
 `RestoredRegistration`; an older provider fails instead of cold-registering.
 
 This adapter composes existing artifact, Codex session, Todo-continuation and
-native-judge owners. It preserves the captured workspace and effective Codex
-configuration, admits a distinct native session and fresh Goal/Agent, and adopts
+native-judge owners. A captured parent directory may contain the native task
+workspace: the receiver resolves exactly one physical task subtree from the
+verified component and restores only that subtree. Captured siblings remain
+inert unless separately selected as research data; absent, ambiguous or
+noncanonical roots reject before the receiver workspace changes.
+
+It preserves the captured workspace and effective Codex configuration, admits
+a distinct native session and fresh Goal/Agent, and adopts
 retained research as evidence on a current preparation Todo. Historical registry,
 claims, leases and settlement records are not installed as current authority.
 If the sealed context contains Explore events, adoption also seeds the fresh

benchmark/runtime/sforge_warm_start.py

@@ -63,10 +63,7 @@ def verify(self, workspace: str) -> dict:
                 or source["goal_id"] == self.goal_id or source["agent_id"] == self.agent_id
                 or hashlib.sha256(self.config.read_bytes()).hexdigest() != native["config_sha256"]):
             raise ValueError("Warm start paths, configuration or destination lineage differ")
-        roots = [c for c in metadata["worker_file_snapshot"]["components"]
-                 if c["source_path"] == workspace]
-        if len(roots) != 1 or not (self.root / "components" / roots[0]["id"]).is_dir():
-            raise ValueError("Warm start requires one complete captured workspace")
+        _workspace_component(self.root, metadata, Path(workspace))
         _research_components(self.root, metadata, Path(workspace), self.research_component_ids)
         context = metadata["research_context"]
         context_path = self.root / "components" / context["component_id"] / context["member_path"]
@@ -116,6 +113,29 @@ async def stage(self, environment, workspace: str):
 _RESEARCH_ROOTS = (Path("/tmp"), Path("/home/workspace"))
 
 
+def _workspace_component(unit: Path, metadata: dict, workspace: Path) -> Path:
+    """Locate the task subtree inside one verified captured file root.
+
+    Capture may seal a parent directory. Only the native workspace is restored;
+    siblings remain inert unless independently selected as research data.
+    Both preflight and materialization use this transport resolution.
+    """
+    matches = []
+    for entry in metadata["worker_file_snapshot"]["components"]:
+        root = Path(entry["source_path"])
+        if workspace.is_relative_to(root):
+            if (not root.is_absolute() or ".." in root.parts or str(root) != entry["source_path"]
+                    or not re.fullmatch(r"[A-Za-z][A-Za-z0-9_.-]{0,63}", entry["id"])):
+                raise ValueError("Captured workspace root must have a canonical path and component ID")
+            matches.append(unit / "components" / entry["id"] / workspace.relative_to(root))
+    if len(matches) != 1:
+        raise ValueError("Warm start requires one complete captured workspace")
+    source = matches[0]
+    if not source.is_dir() or source.resolve() != source:
+        raise ValueError("Captured workspace must be a physical directory inside its component")
+    return source
+
+
 def _research_components(unit: Path, metadata: dict, workspace: Path, selected) -> list[tuple[str, Path, Path]]:
     """Bind explicit receiver selection to sealed worker files, not prose refs.
 
@@ -184,11 +204,7 @@ def restore_workspace(declaration: dict, workspace: Path):
             or unit.is_relative_to(workspace) or Path(_SRC).is_relative_to(workspace)
             or Path(_CODEX_HOME).is_relative_to(workspace)):
         raise ValueError("Checkpoint workspace overlaps the receiver's isolated control paths")
-    roots = [c for c in metadata["worker_file_snapshot"]["components"]
-             if c["source_path"] == str(workspace)]
-    if len(roots) != 1:
-        raise ValueError("Checkpoint must declare exactly one full workspace")
-    source = unit / "components" / roots[0]["id"]
+    source = _workspace_component(unit, metadata, workspace)
     # Reserve the original task-image directory rather than merging stale image
     # files into the captured research state. A partial attempt fails closed.
     original = workspace.with_name(workspace.name + ".before-checkpoint")

benchmark/tests/test_sforge_warm_start.py

@@ -10,7 +10,7 @@
 import pytest
 
 from benchmark.runtime import sforge_warm_start as warm
-from loopx.control_plane.effect_runtime import effect_runtime_result
+from loopx.control_plane.effect_runtime import EffectRuntimeConflict, effect_runtime_result
 
 
 @pytest.mark.parametrize("extension", ["-1", "64800", str(2 ** 53)])
@@ -549,3 +549,80 @@ def test_omitted_fresh_replan_keeps_default_native_history(tmp_path):
     declaration = tmp_path / "warm.json"
     declaration.write_text("{}")
     assert prepare_session_refresh({"LOOPX_WARM_START": str(declaration)}, None, tmp_path) == ""
+
+
+@pytest.fixture
+def parent_workspace_point(point, tmp_path, request):
+    start, original = point
+    parent = tmp_path / "captured-parent"
+    workspace = parent / "nested" / "task"
+    workspace.parent.mkdir(parents=True)
+    shutil.copytree(original, workspace)
+    (parent / "sibling-report").write_text("inert captured sibling")
+    manifest = json.loads((start.root / "checkpoint.json").read_bytes())
+    metadata = manifest["point_metadata"]
+    metadata["native_session"]["guest_workspace"] = str(workspace)
+    component = {"id": "workspace", "source_path": str(parent)}
+    metadata["worker_file_snapshot"]["components"] = [component]
+    variant = getattr(request, "param", "valid")
+    if variant == "ambiguous":
+        metadata["worker_file_snapshot"]["components"].append(
+            {"id": "context", "source_path": str(workspace.parent)})
+    elif variant == "missing":
+        metadata["native_session"]["guest_workspace"] = str(parent / "absent")
+        workspace = parent / "absent"
+    elif variant == "noncanonical":
+        metadata["worker_file_snapshot"]["components"] = [
+            {"id": "workspace", "source_path": str(parent) + "/"}]
+    components = [component, manifest["components"][1]]
+    observed = effect_runtime_result("runtime.checkpoint_artifacts", {"action": "probe", "components": components})
+    unit = tmp_path / "parent-point"
+    sealed = effect_runtime_result("runtime.checkpoint_artifacts", {"action": "capture", "components": components,
+        "point_metadata": metadata, "expected_source_sha256": observed["source_sha256"], "destination": str(unit)})
+    shutil.rmtree(parent)
+    workspace.mkdir(parents=True)
+    (workspace / "candidate").write_text("fresh image candidate")
+    (parent / "sibling-report").write_text("receiver sibling must survive")
+    return replace(start, root=unit, sha256=sealed["sha256"]), workspace, parent
+
+
+def test_parent_capture_restores_only_native_workspace_and_preserves_point(parent_workspace_point):
+    start, workspace, parent = parent_workspace_point
+    assert start.verify(str(workspace))
+    declaration = {"root": str(start.root), "sha256": start.sha256, "deadline_epoch": start.deadline_epoch}
+    warm.restore_workspace(declaration, workspace)
+    assert (workspace / "candidate").read_text() == "retained candidate"
+    assert (workspace / "candidate").stat().st_mode & 0o777 == 0o600
+    assert (workspace.with_name("task.before-checkpoint") / "candidate").read_text() == "fresh image candidate"
+    assert (parent / "sibling-report").read_text() == "receiver sibling must survive"
+    assert not (workspace / "sibling-report").exists()
+    effect_runtime_result("runtime.checkpoint_artifacts", {"action": "verify", "root": str(start.root),
+                                                          "expected_sha256": start.sha256})
+    with pytest.raises(ValueError, match="already prepared"):
+        warm.restore_workspace(declaration, workspace)
+
+
+@pytest.mark.parametrize("parent_workspace_point", ["ambiguous", "missing", "noncanonical"], indirect=True)
+def test_invalid_parent_capture_refuses_preflight_and_restore_before_mutation(parent_workspace_point):
+    start, workspace, parent = parent_workspace_point
+    with pytest.raises(ValueError):
+        start.verify(str(workspace))
+    with pytest.raises(ValueError):
+        warm.restore_workspace({"root": str(start.root), "sha256": start.sha256,
+                                "deadline_epoch": start.deadline_epoch}, workspace)
+    assert (workspace / "candidate").read_text() == "fresh image candidate"
+    assert (parent / "sibling-report").read_text() == "receiver sibling must survive"
+    assert not workspace.with_name(workspace.name + ".before-checkpoint").exists()
+
+
+def test_parent_capture_tamper_refuses_before_receiver_mutation(parent_workspace_point):
+    start, workspace, _ = parent_workspace_point
+    candidate = start.root / "components/workspace/nested/task/candidate"
+    candidate.chmod(0o600)
+    candidate.write_text("tampered")
+    with pytest.raises(EffectRuntimeConflict, match="artifact members differ"):
+        start.verify(str(workspace))
+    with pytest.raises(EffectRuntimeConflict, match="artifact members differ"):
+        warm.restore_workspace({"root": str(start.root), "sha256": start.sha256,
+                                "deadline_epoch": start.deadline_epoch}, workspace)
+    assert (workspace / "candidate").read_text() == "fresh image candidate"