fix(runtime): separate execution admission from public disclosure checks

loopx/cli_commands/turn_decision.py

@@ -38,6 +38,7 @@
     scheduler_execution_context_for_turn,
 )
 from ..status import AUTONOMOUS_REPLAN_PERIODIC_LOOKBACK, collect_status
+from ..paths import default_public_scan_root
 from ..control_plane.effect_runtime import effect_runtime_result
 from .lark_inbox import build_lark_operator_inbox_urgency_projector
 from .turn_selection import turn_controller_advisory_primary
@@ -337,7 +338,10 @@ def current_heartbeat_host_launch_decision(
     args = argparse.Namespace(
         goal_id=goal_id, agent_id=agent_id, turn_instance_id=original_turn,
         todo_id=None, codex_app=False, runtime_profile="generic_cli",
-        scan_path=[str(project)], scan_root=str(project), limit=0,
+        # Uncaptured wakes use heartbeat-prompt's public-package default. Keep
+        # the existing workspace recheck for an already-held checkpoint packet.
+        scan_path=[str(project)] if original_quota is not None else [],
+        scan_root=default_public_scan_root(), limit=0,
         available_capabilities=["shell", "filesystem_write"],
     )
     selection = load_requested_quota_action_selection(

tests/control_plane/test_benchmark_checkpoint_admission.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import json
+from ipaddress import IPv4Address
 import os
 from pathlib import Path
 import sys
@@ -115,7 +116,7 @@ def rendered(command):
 
 
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
-@pytest.mark.parametrize("refusal", ["source_compute_zero", "missing_state", "stop", "original_event_changed", None])
+@pytest.mark.parametrize("refusal", ["source_compute_zero", "missing_state", "stop", "original_event_changed", "public_scope_violation", None])
 def test_checkpoint_resume_rechecks_current_source_with_original_guard(tmp_path, monkeypatch, provider, refusal):
     _run_checkpoint(tmp_path, monkeypatch, provider, refusal=refusal, todo_bound=False)
 
@@ -186,6 +187,13 @@ def captured_then_changed(held, deadline):
             elif refusal == "missing_state":
                 state.rename(state.with_suffix(".saved"))
                 assert _quota(source, runtime, project)["status_health_ok"] is False
+            elif refusal == "public_scope_violation":
+                # The held guard explicitly scanned this workspace. A default
+                # package scan must not replace that check on continuation.
+                (project / "changed_public_fixture.py").write_text(
+                    "endpoint = " + repr("http://" + str(IPv4Address(0x0A000001))) + "\n"
+                )
+                assert _quota(source, runtime, project)["status_health_ok"] is False
             elif refusal == "stop":
                 with monkeypatch.context() as fault:
                     fault.setattr(activation_service, "sync_project_registry_to_global", lambda **kwargs: {
@@ -237,6 +245,7 @@ def captured_then_changed(held, deadline):
         assert error is not None and {
             "source_compute_zero":"goal_launch_not_allowed", "missing_state":"goal_launch_not_allowed",
             "stop":"goal_stopped", "original_event_changed":"heartbeat_launch_original_binding_changed",
+            "public_scope_violation":"goal_launch_not_allowed",
         }[refusal] in str(error)
         assert not (project / "checkpoint-business.json").exists()
         assert fixture._heartbeat_receipt_events(runtime, HELD_TURN) == observations["events"]

tests/control_plane/test_benchmark_current_admission.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import json
+from ipaddress import IPv4Address
 from pathlib import Path
 import sys
 import time
@@ -11,6 +12,7 @@
 from benchmark.runtime import worker
 from benchmark.tests.test_shared_codex_runtime import worker_env
 from loopx.configure_goal import configure_goal
+from loopx.cli_commands import turn_decision
 from loopx.control_plane.goals import activation_service
 from loopx.control_plane.goals.first_party_host_admission import FirstPartyHostRuntimeRejected
 from loopx.global_registry import sync_project_registry_to_global
@@ -79,6 +81,12 @@ def _quota(registry, runtime, project):
 @pytest.mark.parametrize("refusal", ["source_compute_zero", "missing_state", "stop", None])
 def test_heartbeat_worker_qualifies_current_source_at_actual_spawn(tmp_path, monkeypatch, provider, refusal):
     env, project, runtime, source, state = _environment(tmp_path, monkeypatch, provider)
+    if refusal is None:
+        # This default-scan caller may work on a private task. Keep that fixture
+        # out of the shared environment used by explicit project-scan callers.
+        (project / "task_network_fixture.py").write_text(
+            "endpoint = " + repr("http://" + str(IPv4Address(0x0A000001))) + "\n"
+        )
     mirror = Path(env["LOOPX_REGISTRY"])
     mirror_before, state_before = mirror.read_bytes(), state.read_bytes()
     observations = []
@@ -112,6 +120,7 @@ def after_session_selection(*arguments):
     monkeypatch.setattr(worker, "native_command", after_session_selection)
     bindings_before = _bindings(runtime)
     rejected = None
+    result = None
     try:
         result = worker.run_once(env)
     except FirstPartyHostRuntimeRejected as error:
@@ -140,6 +149,9 @@ def test_initial_benchmark_planner_can_launch_without_a_prior_heartbeat_receipt(
     env, project, runtime, source, _state = _environment(tmp_path, monkeypatch, provider, replan="empty")
     packet = _quota(source, runtime, project)
     assert packet["should_run"] is True and packet.get("selected_todo") is None
+    (project / "task_network_fixture.py").write_text(
+        "endpoint = " + repr("http://" + str(IPv4Address(0x0A000001))) + "\n"
+    )
     env.update(LOOPX_TASK_ENTRY="loopx-planned", LOOPX_TASK_STAGE="plan", LOOPX_EXECUTION_MODE=mode,
         LOOPX_PHASE_DEADLINE_EPOCH=str(time.time()+600), LOOPX_PLANNING_RESULT=str(project / "planning.json"))
     if mode == "turn":
@@ -166,3 +178,24 @@ def test_initial_benchmark_planner_can_launch_without_a_prior_heartbeat_receipt(
     assert len(planned["todo_ids"]) == 1 and result["planning"]["todo_ids"] == planned["todo_ids"]
     assert not fixture._heartbeat_receipt_events(runtime, result["turn_id"])
     assert fixture._spend_run_count(runtime) == 0
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_heartbeat_host_launch_retains_public_scan_refusal(tmp_path, monkeypatch, provider):
+    env, project, runtime, source, _state = _environment(tmp_path, monkeypatch, provider)
+    public = tmp_path / "selected-public"
+    public.mkdir()
+    (public / "unsafe_fixture.py").write_text(
+        "endpoint = " + repr("http://" + str(IPv4Address(0x0A000001))) + "\n"
+    )
+    code, original = fixture._run_cli(source, runtime,
+        "quota", "should-run", "--goal-id", fixture.GOAL_ID, "--agent-id", fixture.AGENT_ID,
+        "--scan-path", str(public), cwd=project)
+    assert code == 1 and original["status_health_ok"] is False and original["should_run"] is False
+    monkeypatch.setattr(turn_decision, "default_public_scan_root", lambda: str(public))
+    current = turn_decision.current_heartbeat_host_launch_decision(
+        registry_path=Path(env["LOOPX_REGISTRY"]), runtime_root=runtime, project=project,
+        goal_id=fixture.GOAL_ID, agent_id=fixture.AGENT_ID, turn_instance_id="public-scan-refusal",
+    )["current"]
+    assert current["status_health_ok"] is False and current["should_run"] is False
+    assert not (project / "native-starts").exists() and not (project / "model-starts").exists()