fix(host): recheck Goal admission before process launch and turn dispatch

benchmark/runtime/checkpoint_wake.py

@@ -184,17 +184,22 @@ def _capture(self, held, deadline):
             }, timeout=max(.1, window - time.time()), retry_safe=False, large_local_snapshot=True)
 
     def run(self, execution, body, wake, session_wake, stdout, stderr, timeout):
-        from .worker import child_process, native_command
+        from .worker import child_process, heartbeat_launch_check, native_command
         from .sessions import BenchmarkSessionWake
 
         deadline = time.monotonic() + timeout if timeout is not None else float("inf")
+        launch_env, original_quota = self.env, None
         while True:
             remaining = deadline - time.monotonic()
             if remaining <= 0:
                 raise subprocess.TimeoutExpired("checkpoint source wake", timeout)
-            command = native_command(self.env, execution, "execute", wake, session_wake)
+            command = native_command(launch_env, execution, "execute", wake, session_wake)
             continuation = None
-            with child_process(command, env=self.env, stdout=stdout, stderr=stderr) as process:
+            with child_process(command, env=launch_env, stdout=stdout, stderr=stderr,
+                               goal_admission=session_wake.admission if session_wake is not None else None,
+                               check_current_launch=heartbeat_launch_check(
+                                   launch_env, session_wake, original_quota=original_quota)) as process:
+                self.receipt["host_invoked"] = True
                 process.stdin.write(body)
                 process.stdin.close()
                 while process.poll() is None:
@@ -252,4 +257,9 @@ def run(self, execution, body, wake, session_wake, stdout, stderr, timeout):
             session_wake = BenchmarkSessionWake(self.env, execution, self.receipt)
             if not source_session_id or session_wake.session_id != source_session_id:
                 raise ValueError("checkpoint source continuation lost its native session")
+            # The held CLI guard can belong to another Turn inside this wake.
+            # Retain its typed packet, never recover identity from prompt text
+            # or reuse the enclosing wake's ID for its remaining settlement.
+            original_quota = held["quota"]
+            launch_env = {**self.env, "LOOPX_TURN": original_quota["heartbeat_receipt"]["turn_instance_id"]}
             body = continuation

benchmark/runtime/worker.py

@@ -10,7 +10,7 @@
 import subprocess
 import time
 import uuid
-from contextlib import ExitStack, contextmanager
+from contextlib import ExitStack, contextmanager, nullcontext
 from dataclasses import replace
 from pathlib import Path
 
@@ -96,21 +96,48 @@ def heartbeat_body(env: dict[str, str], turn_id: str, *, native_goal: bool,
 
 
 @contextmanager
-def child_process(command: list[str], *, env: dict[str, str], stdout, stderr):
-    process = subprocess.Popen(
-        command,
-        cwd=env["LOOPX_PROJECT"],
-        env=env,
-        stdin=subprocess.PIPE,
-        stdout=stdout,
-        stderr=stderr,
-        text=True,
-        start_new_session=True,
-    )
+def child_process(command: list[str], *, env: dict[str, str], stdout, stderr,
+                  goal_admission=None, check_current_launch=None):
+    process = None
     try:
+        if check_current_launch is not None and goal_admission is None:
+            raise ValueError("current Host launch check requires its original Goal admission")
+        if check_current_launch is not None:
+            goal_admission = goal_admission.with_launch_decision(check_current_launch)
+        # Session selection may precede rendering/capture. Serialize only actual
+        # launch with the existing Goal lifetime owner, never the whole model run.
+        with (goal_admission.launch_lifetime(operation="codex_agent_host_launch")
+              if goal_admission is not None else nullcontext()):
+            process = subprocess.Popen(
+                command,
+                cwd=env["LOOPX_PROJECT"],
+                env=env,
+                stdin=subprocess.PIPE,
+                stdout=stdout,
+                stderr=stderr,
+                text=True,
+                start_new_session=True,
+            )
         yield process
     finally:
-        _reap_child_process(process)
+        if process is not None:
+            _reap_child_process(process)
+
+
+def heartbeat_launch_check(env, session_wake, *, original_quota=None):
+    """Carry the Host's real Turn inputs into the product's current quota owner."""
+    if session_wake is None:
+        return None  # Plain/native-Goal transports have no LoopX heartbeat owner.
+    from loopx.cli_commands.turn_decision import current_heartbeat_host_launch_decision
+
+    def check(registry_path):
+        return current_heartbeat_host_launch_decision(
+            registry_path=registry_path, runtime_root=Path(env["LOOPX_RUNTIME_ROOT"]),
+            project=Path(env["LOOPX_PROJECT"]), goal_id=env["LOOPX_GOAL_ID"],
+            agent_id=env["LOOPX_AGENT_ID"], turn_instance_id=env["LOOPX_TURN"],
+            original_quota=original_quota, goal_ref=session_wake.goal_ref,
+        )
+    return check
 
 
 def _reap_child_process(process):
@@ -435,6 +462,7 @@ def run_once(env: dict[str, str]) -> dict:
             from benchmark.runtime.sessions import BenchmarkSessionWake
 
             session_wake = BenchmarkSessionWake(env, execution, receipt)
+            receipt["host_invoked"] = False
             if stage == "execute" and execution.mode == "heartbeat" and env.get("LOOPX_WARM_START"):
                 from benchmark.runtime.checkpoint_session_refresh import prepare_session_refresh
 
@@ -478,7 +506,11 @@ def run_once(env: dict[str, str]) -> dict:
                             execution, body, wake, session_wake, stdout, stderr, timeout)
                         receipt["ok"] = receipt["return_code"] == 0
                     else:
-                        with child_process(command, env=env, stdout=stdout, stderr=stderr) as process:
+                        with child_process(command, env=env, stdout=stdout, stderr=stderr,
+                                goal_admission=session_wake.admission if session_wake is not None else None,
+                                check_current_launch=heartbeat_launch_check(env, session_wake)) as process:
+                            if session_wake is not None:
+                                receipt["host_invoked"] = True
                             process.communicate(input=body, timeout=timeout)
                             receipt["return_code"] = process.returncode
                             receipt["ok"] = process.returncode == 0
@@ -516,7 +548,7 @@ def run_once(env: dict[str, str]) -> dict:
     finally:
         checkpoint_context.close()
         try:
-            if session_wake is not None:
+            if session_wake is not None and receipt.get("host_invoked") is True:
                 session_wake.observe(wake / "stdout.jsonl")
         except BaseException as exc:
             receipt.update(ok=False, **_error_tail(exc))

benchmark/tests/test_resume_sessions.py

@@ -16,6 +16,48 @@
 from loopx.control_plane.turn_driver.codex_sessions import load_codex_cli_session
 
 
+@pytest.mark.skipif(os.name != "posix", reason="POSIX owned Host process groups")
+def test_launch_admission_exit_interrupt_reaps_started_host(tmp_path, monkeypatch):
+    from contextlib import contextmanager
+    import subprocess
+    import sys
+
+    from benchmark.runtime import worker
+
+    class Admission:
+        @contextmanager
+        def launch_lifetime(self, *, operation):
+            yield
+            raise KeyboardInterrupt("interrupted while releasing launch admission")
+
+    children = []
+    popen = subprocess.Popen
+
+    def record_child(*args, **kwargs):
+        process = popen(*args, **kwargs)
+        children.append(process)
+        return process
+
+    monkeypatch.setattr(worker.subprocess, "Popen", record_child)
+    try:
+        with pytest.raises(KeyboardInterrupt, match="releasing launch admission"):
+            with worker.child_process(
+                [sys.executable, "-c", "import time; time.sleep(60)"],
+                env={**os.environ, "LOOPX_PROJECT": str(tmp_path)},
+                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+                goal_admission=Admission(),
+            ):
+                pytest.fail("interrupted admission must not yield the Host")
+        assert len(children) == 1
+        assert children[0].poll() is not None
+        with pytest.raises(ProcessLookupError):
+            os.killpg(children[0].pid, 0)
+    finally:
+        for process in children:
+            worker._reap_child_process(process)
+            process.stdin.close()
+
+
 @pytest.mark.parametrize("existing", ["valid", "malformed", "dangling"])
 def test_absent_only_session_admission_preserves_every_existing_binding(tmp_path, existing):
     """The no-replacement condition is checked under the session owner's lock."""
@@ -88,10 +130,21 @@ def resume_env(tmp_path, monkeypatch):
         LOOPX_SHARED_SKILLS=str(skills),
         LOOPX_CLI="loopx",
     )
+    Path(env["LOOPX_REGISTRY"]).write_text(json.dumps({
+        "schema_version": "0.2", "projects": [], "goals": [{
+            "id": env["LOOPX_GOAL_ID"], "status": "active", "repo": str(tmp_path),
+            "coordination": {"registered_agents": [env["LOOPX_AGENT_ID"]]},
+        }],
+    }))
     monkeypatch.setattr(
         "benchmark.runtime.worker.heartbeat_body",
         lambda *a, **kw: "Current Todo input.",
     )
+    # This fixture isolates native conversation/lifetime transport, including
+    # deliberately unregistered identities and lifecycle-only source profiles.
+    # Real current quota/health qualification is exercised by the CLI-backed
+    # test_benchmark_current_admission and checkpoint_admission integration tests.
+    monkeypatch.setattr("benchmark.runtime.worker.heartbeat_launch_check", lambda *a, **kw: None)
     Path(env["CODEX_BIN"]).write_text(
         f"#!{__import__('sys').executable}\n"
         + """
@@ -176,8 +229,10 @@ def test_timeout_keeps_the_observed_session_for_next_heartbeat(tmp_path, monkeyp
     observed_sessions = []
 
     @contextmanager
-    def timeout_after_session_event(command, *, env, stdout, stderr):
-        with real_child_process(command, env=env, stdout=stdout, stderr=stderr) as process:
+    def timeout_after_session_event(command, *, env, stdout, stderr, goal_admission=None,
+                                    check_current_launch=None):
+        with real_child_process(command, env=env, stdout=stdout, stderr=stderr,
+                goal_admission=goal_admission, check_current_launch=check_current_launch) as process:
             def communicate(*, input, timeout):
                 assert timeout == 1
                 # Inject the timeout after the real child publishes its native
@@ -208,6 +263,57 @@ def communicate(*, input, timeout):
     assert second["session"]["session_id"] == first["session"]["session_id"]
 
 
+@pytest.mark.parametrize("change", ["stop", "replace"])
+@pytest.mark.parametrize("profile", ["legacy", "source"])
+def test_heartbeat_rechecks_selected_session_at_native_launch(tmp_path, monkeypatch, change, profile):
+    from benchmark.runtime import worker
+    from loopx.control_plane.goals.first_party_host_admission import FirstPartyHostRuntimeRejected
+    from loopx.control_plane.projects.registry_codec import project_registry_transaction, source_session_registry_transaction
+    from tests.test_loopx_turn_codex_cli import _source_admission
+
+    env = resume_env(tmp_path, monkeypatch)
+    if profile == "source":
+        admission = _source_admission(tmp_path)
+        env["LOOPX_REGISTRY"] = str(admission.registry_path)
+    registry_path = Path(env["LOOPX_REGISTRY"])
+    registry_transaction = source_session_registry_transaction if profile == "source" else project_registry_transaction
+    first = run_once(env)
+    original_binding = binding(env)
+    history = next((tmp_path / "logs/sessions").glob("*.jsonl"))
+    original_history = history.read_bytes()
+    native_command = worker.native_command
+
+    def changed_after_selection(*args, **kwargs):
+        command = native_command(*args, **kwargs)
+        with registry_transaction(
+            registry_path, operation="fixture_stop_before_native_launch",
+        ) as transaction:
+            registry = transaction.payload_copy()
+            if change == "replace":
+                if profile == "source":
+                    registry["goals"][0]["goal_instance_id"] = "ginst_" + "b" * 32
+                else:
+                    registry["goals"][0]["repo"] = str(tmp_path / "replacement")
+            else:
+                registry["goals"][0]["activation"] = {
+                    "schema_version": "loopx_goal_activation_v1", "state": "stopped",
+                    "updated_at": "2026-10-10T12:00:00Z", "reason": "Owner stopped this Goal",
+                }
+            transaction.commit(registry)
+        return command
+
+    monkeypatch.setattr(worker, "native_command", changed_after_selection)
+    with pytest.raises(FirstPartyHostRuntimeRejected) as error:
+        run_once(env)
+    assert error.value.code == ("goal_stopped" if change == "stop" else
+                                "stale_goal_instance" if profile == "source" else "stale_goal_authority")
+    receipt = json.loads(sorted((tmp_path / "logs/wakes").glob("*/receipt.json"))[-1].read_bytes())
+    assert receipt["host_invoked"] is False and not receipt["ok"]
+    assert receipt["error_kind"] == "FirstPartyHostRuntimeRejected"
+    assert receipt["session"]["session_id"] == first["session"]["session_id"]
+    assert binding(env) == original_binding and history.read_bytes() == original_history
+
+
 @pytest.mark.parametrize("mutation", ["corrupt", "home", "fork"])
 def test_resume_refuses_silent_replacement(tmp_path, monkeypatch, mutation):
     env = resume_env(tmp_path, monkeypatch)
@@ -236,6 +342,13 @@ def test_another_identity_starts_an_independent_conversation(
 ):
     env = resume_env(tmp_path, monkeypatch)
     first = run_once(env)
+    if field == "LOOPX_GOAL_ID":
+        from loopx.control_plane.projects.registry_codec import project_registry_transaction
+
+        with project_registry_transaction(Path(env["LOOPX_REGISTRY"]), operation="fixture_second_goal") as transaction:
+            payload = transaction.payload_copy()
+            payload["goals"].append({**payload["goals"][0], "id": "another-identity"})
+            transaction.commit(payload)
     second = run_once(env | {field: "another-identity"})
     assert second["session"]["action"] == "start_new"
     assert second["session"]["session_id"] != first["session"]["session_id"]

benchmark/tests/test_task_entry.py

@@ -70,8 +70,10 @@ def planning_env(tmp_path):
                     {
                         "id": "planning-goal",
                         "repo": str(project),
+                        "domain": "project",
                         "state_file": str(state),
                         "status": "active",
+                        "adapter": {"kind": "read_only_project_map_v0", "status": "connected-read-only"},
                         "coordination": {"registered_agents": ["planner"]},
                     }
                 ],