Validate checkpoint capture routes before quota admission

benchmark/runtime/checkpoint_wake.py

@@ -18,6 +18,7 @@
 from pathlib import Path
 
 from loopx.control_plane.effect_runtime import effect_runtime_result
+from loopx.control_plane.turn_driver.decision_capture import bind_handoff_capture_root
 from .checkpoint import _mark_replan_attempt, _participant_id, held_quota_decision, read_checkpoint_request
 from .checkpoint_handoff import CHECKPOINT_CAPTURE_SECONDS
 
@@ -55,6 +56,7 @@ def __enter__(self):
             self.listener = self.stack.enter_context(socket.socket(socket.AF_UNIX, socket.SOCK_STREAM))
             self.listener.bind(self.socket_path)
             self.listener.listen(1)
+            bind_handoff_capture_root(self.socket_path, self.capture_root)
             return self
         except BaseException:
             self.stack.close()

benchmark/tests/test_checkpoint_handoff.py

@@ -295,6 +295,61 @@ def controller(root, components, metadata, probe, deadline, **kwargs):
     assert code.read_text() == "retained source"
 
 
+@pytest.mark.skipif(os.name != "posix", reason="Requires the Host Unix rendezvous")
+def test_foreign_capture_root_does_not_commit_quota_guard(tmp_path):
+    from pathlib import Path
+    import subprocess
+
+    from benchmark.runtime.checkpoint import held_quota_decision
+    from benchmark.runtime.checkpoint_wake import CheckpointWake
+    from tests.control_plane.test_quota_settlement_cli import (
+        AGENT_ID, GOAL_ID, TURN_ID, _configure_autonomous_replan_fixture,
+        _heartbeat_receipt_count, _initialize_fixture_authority, _write_fixture,
+    )
+
+    project, runtime, registry = _write_fixture(tmp_path / "authority")
+    _configure_autonomous_replan_fixture(project, runtime, registry)
+    _initialize_fixture_authority(project, runtime, provider="sqlite")
+    request = tmp_path / "checkpoint-request.json"
+    request.write_text(json.dumps({
+        "destination": str(tmp_path / "checkpoints"), "trigger": "fresh_replan",
+        "point_metadata": {}, "components": [{"id": "workspace", "source_path": str(project)}],
+    }))
+    wake_env = {
+        "LOOPX_WAKE_LOG_DIR": str(tmp_path / "wakes"),
+        "LOOPX_CHECKPOINT_REQUEST": str(request),
+    }
+    with CheckpointWake(wake_env, {"mode": "heartbeat", "stage": "execute", "turn_id": TURN_ID}) as host:
+        assert host.listener is not None
+        host.capture_root.mkdir(parents=True, mode=0o700)
+        native_context = tmp_path / "native-context"
+        native_context.mkdir(mode=0o700)
+        command = [sys.executable, "-m", "loopx.entrypoint", "--format", "json",
+                   "--registry", str(registry), "--runtime-root", str(runtime), "quota", "should-run",
+                   "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, "--turn-instance-id", TURN_ID,
+                   "--runtime-profile", "generic_cli", "--context-delivery", "off",
+                   "--decision-handoff-socket", host.socket_path, "--decision-output-root"]
+        env = {**os.environ, "PYTHONPATH": str(Path(__file__).resolve().parents[2])}
+        refused = subprocess.run([*command, str(native_context)], cwd=project, env=env,
+                                 capture_output=True, text=True, timeout=15, check=False)
+        assert refused.returncode == 1
+        assert "capture root differs from the current Host wake" in refused.stdout + refused.stderr
+        assert not list(native_context.iterdir())
+        assert _heartbeat_receipt_count(runtime, TURN_ID) == 0
+
+        host.listener.settimeout(15)
+        with subprocess.Popen([*command, str(host.capture_root)], cwd=project, env=env,
+                              stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) as process:
+            with held_quota_decision(host.listener, host.capture_root, goal_id=GOAL_ID,
+                                    agent_id=AGENT_ID, turn_instance_id=TURN_ID,
+                                    interrupt_when=lambda _: False) as held:
+                assert held["quota"]["ok"] is True
+                assert held["path"].parent.parent == host.capture_root
+            stdout, stderr = process.communicate(timeout=15)
+            assert process.returncode == 0, (stdout, stderr)
+    assert _heartbeat_receipt_count(runtime, TURN_ID) == 1
+
+
 @pytest.mark.skipif(sys.platform != "linux", reason="Qualifies the Linux writer process group")
 @pytest.mark.parametrize("use_selector,closeout,settlement_only", [
     (False, False, False), (True, False, False), (True, True, False), (True, False, True),

loopx/cli_commands/quota_capture.py

@@ -510,8 +510,13 @@ def prepare_decision_capture(args: argparse.Namespace) -> Path | None:
         raise QuotaCommandValidationError(
             "native Codex context delivery needs JSON and a capture destination")
     if handoff is not None:
-        validate_handoff_socket(handoff)
-        if requested is None and root is None:
+        handoff_root = None
+        if root is not None:
+            handoff_root = Path(root).expanduser().absolute()
+        elif requested is not None:
+            handoff_root = Path(requested).expanduser().absolute().parent
+        validate_handoff_socket(handoff, capture_root=handoff_root)
+        if handoff_root is None:
             raise QuotaCommandValidationError("decision handoff requires an explicit decision capture destination")
     if requested is None and root is None:
         return None

loopx/control_plane/heartbeat/host.py

@@ -58,7 +58,7 @@ def resolve_heartbeat_capture_root(
     if decision_handoff_socket is not None:
         from ..turn_driver.decision_capture import validate_handoff_socket
 
-        validate_handoff_socket(decision_handoff_socket)
+        validate_handoff_socket(decision_handoff_socket, capture_root=directory)
     return directory
 
 

loopx/control_plane/turn_driver/decision_capture.py

@@ -229,7 +229,45 @@ def prepare_turn_decision_capture(
         decision=decision, existing_journal=existing_journal).publish()
 
 
-def validate_handoff_socket(value: str) -> Path:
+_HANDOFF_CAPTURE_ROOT_LIMIT = 4096
+
+
+def _handoff_capture_root_file(endpoint: Path) -> Path:
+    return endpoint.with_name(endpoint.name + ".capture-root")
+
+
+def bind_handoff_capture_root(socket_path: str, capture_root: Path) -> None:
+    """Publish the current Host's transport destination beside its private socket."""
+    endpoint = validate_handoff_socket(socket_path)
+    raw = os.fsencode(capture_root)
+    if capture_root.resolve() != capture_root or len(raw) > _HANDOFF_CAPTURE_ROOT_LIMIT:
+        raise ValueError("decision handoff requires a physical absolute capture root")
+    binding = _handoff_capture_root_file(endpoint)
+    with os.fdopen(os.open(binding, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600), "wb") as stream:
+        stream.write(raw)
+
+
+def _validate_handoff_capture_root(endpoint: Path, requested_root: Path) -> None:
+    try:
+        with os.fdopen(open_regular_capture(_handoff_capture_root_file(endpoint)), "rb") as stream:
+            metadata = os.fstat(stream.fileno())
+            if metadata.st_uid != os.getuid() or metadata.st_mode & 0o077:
+                raise ValueError("capture root binding must be private to the Host user")
+            bound_root = stream.read(_HANDOFF_CAPTURE_ROOT_LIMIT + 1)
+            if not bound_root or len(bound_root) > _HANDOFF_CAPTURE_ROOT_LIMIT:
+                raise ValueError("invalid capture root binding size")
+    except FileNotFoundError:
+        return  # Earlier Hosts still validate the capture when accepting it.
+    except (OSError, ValueError) as exc:
+        raise QuotaCommandValidationError(f"invalid decision handoff binding; no guard was run: {exc}") from exc
+    if bound_root != os.fsencode(requested_root):
+        raise QuotaCommandValidationError(
+            "decision handoff capture root differs from the current Host wake; no guard was run. "
+            "Use the --decision-output-root named by the current wake."
+        )
+
+
+def validate_handoff_socket(value: str, *, capture_root: Path | None = None) -> Path:
     """A caller-owned local transport, not an authority or admission endpoint."""
     path = Path(value)
     try:
@@ -250,6 +288,8 @@ def validate_handoff_socket(value: str) -> Path:
         ) from exc
     except OSError as exc:
         raise QuotaCommandValidationError(f"invalid decision handoff socket: {exc}") from exc
+    if capture_root is not None:
+        _validate_handoff_capture_root(path, capture_root)
     return path