feat(benchmark): enforce request admission at provider dispatch

benchmark/runtime/checkpoint_wake.py

@@ -183,7 +183,7 @@ def _capture(self, held, deadline):
                 "deadline_epoch_ms": window * 1000,
             }, timeout=max(.1, window - time.time()), retry_safe=False, large_local_snapshot=True)
 
-    def run(self, execution, body, wake, session_wake, stdout, stderr, timeout):
+    def run(self, execution, body, wake, session_wake, stdout, stderr, timeout, *, request_guard=None):
         from .worker import child_process, heartbeat_launch_check, native_command
         from .sessions import BenchmarkSessionWake
 
@@ -194,12 +194,16 @@ def run(self, execution, body, wake, session_wake, stdout, stderr, timeout):
             if remaining <= 0:
                 raise subprocess.TimeoutExpired("checkpoint source wake", timeout)
             command = native_command(launch_env, execution, "execute", wake, session_wake)
+            if request_guard is not None:
+                command = request_guard.command(command)
             continuation = None
             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
+                if request_guard is not None:
+                    request_guard.bind(process, launch_env, session_wake, original_quota=original_quota)
                 process.stdin.write(body)
                 process.stdin.close()
                 while process.poll() is None:

benchmark/runtime/research_request_guard.py

@@ -0,0 +1,136 @@
+"""Bind each request in a research wake to the existing live launch owner.
+
+This is transport instrumentation, not an alternative quota/wait decision. No
+prompt, response body, credential, or upstream URL is written to the journal.
+"""
+from __future__ import annotations
+
+from contextlib import ExitStack, contextmanager
+import json
+import os
+from pathlib import Path
+import signal
+import threading
+import time
+from urllib.parse import urlsplit
+from urllib.request import proxy_bypass_environment
+
+from loopx.capabilities.benchmark_toolkit.provider_gateway import (
+    ProviderGatewayRequestRejected, serve_runner_owned_provider_gateway,
+)
+from loopx.control_plane.goals.first_party_host_admission import FirstPartyHostRuntimeRejected
+
+
+class ResearchRequestGuard:
+    def __init__(self, env, receipt, wake):
+        if not env.get("OPENAI_BASE_URL") or not env.get("OPENAI_API_KEY"):
+            raise ValueError("request admission requires the declared custom model endpoint")
+        if env.get("CODEX_WIRE_API", "responses") != "responses":
+            raise ValueError("request admission currently requires the Responses transport")
+        self.env, self.receipt, self.wake = env, receipt, Path(wake)
+        self.process = None
+        self.admission = None
+        self.refused = False
+        self.admission_error = False
+        self.lock = threading.Lock()
+        self.binding = {}
+        self.base_url = None
+
+    def bind(self, process, env, session_wake, original_quota=None):
+        from .worker import heartbeat_launch_check
+
+        if session_wake is None:
+            raise ValueError("research request admission requires the original session owner")
+        with self.lock:
+            self.process = process
+            self.admission = session_wake.admission.with_launch_decision(
+                heartbeat_launch_check(env, session_wake, original_quota=original_quota))
+            self.binding = {"wake_id": self.receipt["turn_id"], "loopx_turn_id": env["LOOPX_TURN"],
+                "goal_id": env["LOOPX_GOAL_ID"], "agent_id": env["LOOPX_AGENT_ID"],
+                "bound_native_session_id": session_wake.session_id}
+
+    def lifetime(self, request):
+        # Capture before the gateway connects. A delayed request must not borrow
+        # a checkpoint continuation's replacement process or held Turn binding.
+        with self.lock:
+            owner, process = self.admission, self.process
+            request.update(self.binding)
+        try:
+            self._require_live_binding(owner, process)
+        except Exception as error:
+            self._reject(request, owner, process, error)
+        return self._bound_lifetime(request, owner, process)
+
+    def _require_live_binding(self, owner, process):
+        with self.lock:
+            if self.refused:
+                raise FirstPartyHostRuntimeRejected("research_original_request_recovery_required")
+            if (owner is None or process is None or self.admission is not owner
+                    or self.process is not process or process.poll() is not None):
+                raise FirstPartyHostRuntimeRejected("research_request_without_live_owner")
+            if time.time() >= float(self.env["LOOPX_PHASE_DEADLINE_EPOCH"]):
+                raise FirstPartyHostRuntimeRejected("research_phase_budget_exhausted")
+
+    def _reject(self, request, owner, process, error):
+        # Failure to read authority is not permission. Only admission calls
+        # reach here; a connect/send error remains a transport observation.
+        code = error.code if isinstance(error, FirstPartyHostRuntimeRejected) else "research_request_admission_unavailable"
+        with self.lock:
+            request["refusal_code"] = code
+            current = self.admission is owner and self.process is process
+            if current:
+                # A late request retains its own rejection evidence, but must
+                # not poison a checkpoint replacement's valid admission. A
+                # genuine current refusal stays latched until the wake returns
+                # to its scheduler; bind() cannot clear that recovery boundary.
+                self.refused = True
+                self.admission_error = self.admission_error or not isinstance(error, FirstPartyHostRuntimeRejected)
+                self.receipt.setdefault("request_admission_refused", {"request_id": request["request_id"], "code": code})
+            if process is not None and (current or self.process is not process) and process.poll() is None:
+                try:
+                    os.killpg(process.pid, signal.SIGINT)
+                except ProcessLookupError:
+                    pass
+        raise ProviderGatewayRequestRejected() from None
+
+    @contextmanager
+    def _bound_lifetime(self, request, owner, process):
+        with ExitStack() as scope:
+            try:
+                self._require_live_binding(owner, process)
+                scope.enter_context(owner.launch_lifetime(operation="benchmark_provider_request"))
+                # Reading canonical admission can wait behind a writer. The
+                # original process, phase clock and refusal must still be live
+                # after that wait and immediately before the provider send.
+                self._require_live_binding(owner, process)
+            except Exception as error:
+                self._reject(request, owner, process, error)
+            yield
+
+    def observe(self, event):
+        with self.lock:
+            with (self.wake / "provider-requests.jsonl").open("a", encoding="utf-8") as stream:
+                stream.write(json.dumps(event, allow_nan=False) + "\n")
+                stream.flush()
+                os.fsync(stream.fileno())
+
+    @contextmanager
+    def serve(self):
+        target = urlsplit(self.env["OPENAI_BASE_URL"])
+        proxy = None
+        if not proxy_bypass_environment(target.netloc, {"no": self.env.get("NO_PROXY", self.env.get("no_proxy", ""))}):
+            key = "HTTPS_PROXY" if target.scheme == "https" else "HTTP_PROXY"
+            proxy = self.env.get(key, self.env.get(key.lower()))
+        with serve_runner_owned_provider_gateway(upstream_base_url=self.env["OPENAI_BASE_URL"],
+                upstream_bearer_token=self.env["OPENAI_API_KEY"],
+                upstream_proxy_url=proxy,
+                request_lifetime=self.lifetime, observe_request=self.observe) as gateway:
+            self.base_url = gateway.base_url
+            yield self
+
+    def command(self, command):
+        if self.base_url is None:
+            raise ValueError("request gateway is not running")
+        # Invocation-only routing; the pinned home and session profile retain
+        # the actual upstream. Restart never persists an obsolete loopback port.
+        return [command[0], "-c", "model_providers.harbor.base_url=" + json.dumps(self.base_url), *command[1:]]

benchmark/runtime/worker.py

@@ -362,6 +362,8 @@ def _error_tail(exc: BaseException) -> dict[str, str]:
 
 
 def run_once(env: dict[str, str]) -> dict:
+    if env.get("LOOPX_RESEARCH_LOOP", "0") not in {"0", "1"}:
+        raise ValueError("LOOPX_RESEARCH_LOOP must be 0 or 1")
     if env.get("LOOPX_TURN_ENVELOPE", "0") not in {"0", "1"}:
         raise ValueError("LOOPX_TURN_ENVELOPE must be 0 or 1")
     execution = Execution(
@@ -389,6 +391,7 @@ def run_once(env: dict[str, str]) -> dict:
         "mode": execution.mode,
         "context": execution.context,
         "task_entry": execution.task_entry,
+        **({"research_loop": True} if env.get("LOOPX_RESEARCH_LOOP") == "1" else {}),
         **({"turn_envelope": True, "turn_envelope_source": "explicit"}
            if execution.turn_envelope else {}),
         "stage": stage,
@@ -397,6 +400,7 @@ def run_once(env: dict[str, str]) -> dict:
         "timed_out": False,
     }
     session_wake = None
+    request_guard = None
     checkpoint_context = ExitStack()
     pending_path = (
         Path(env.get("LOOPX_RUNTIME_ROOT", str(home))) / "benchmark-pending-turn.json"
@@ -466,6 +470,12 @@ def run_once(env: dict[str, str]) -> dict:
             from benchmark.runtime.checkpoint import capture_before_replan
 
             body = capture_before_replan(env, receipt, body)
+        if env.get("LOOPX_RESEARCH_LOOP") == "1":
+            from .research_request_guard import ResearchRequestGuard
+
+            if execution.mode != "heartbeat":
+                raise ValueError("research request guard requires heartbeat execution")
+            request_guard = checkpoint_context.enter_context(ResearchRequestGuard(env, receipt, wake).serve())
         with (wake / "stderr.log").open("w") as stderr:
             if execution.native_goal and stage == "execute":
                 run_native_goal(env, execution, body, receipt, stderr)
@@ -498,17 +508,27 @@ def run_once(env: dict[str, str]) -> dict:
                                    else execution.timeout_seconds + allowance)
                     if checkpoint_wake.listener is not None:
                         receipt["return_code"] = checkpoint_wake.run(
-                            execution, body, wake, session_wake, stdout, stderr, timeout)
+                            execution, body, wake, session_wake, stdout, stderr, timeout,
+                            **({"request_guard": request_guard} if request_guard is not None else {}))
                         receipt["ok"] = receipt["return_code"] == 0
                     else:
+                        if request_guard is not None:
+                            command = request_guard.command(command)
                         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
+                            if request_guard is not None:
+                                request_guard.bind(process, env, session_wake)
                             process.communicate(input=body, timeout=timeout)
                             receipt["return_code"] = process.returncode
                             receipt["ok"] = process.returncode == 0
+                if request_guard is not None and request_guard.refused and stage == "execute":
+                    # The wake returns to its original scheduler for a current
+                    # wait/stop/recovery read; native interruption is not work
+                    # success and never manufactures a settlement.
+                    receipt["ok"] = not request_guard.admission_error
                 if stage == "plan" and receipt["ok"]:
                     from benchmark.runtime.planning import task_plan_packet, validate_plan_readback
 

benchmark/tests/test_task_entry.py

@@ -177,6 +177,37 @@ def test_planning_writes_real_todo_then_reuses_it_without_executing_task(
     )
 
 
+def test_research_worker_routes_each_resume_without_changing_the_pinned_home(planning_env):
+    from tests.capabilities.test_provider_gateway import _recording_upstream
+
+    binary = Path(planning_env["CODEX_BIN"])
+    request = '''import urllib.request
+endpoint = json.loads(next(value.split("=", 1)[1] for value in sys.argv if value.startswith("model_providers.harbor.base_url=")))
+with urllib.request.urlopen(urllib.request.Request(endpoint + "/responses", data=b"{}", headers={
+    "session-id": "planning-fixture-session", "x-client-request-id": "fixture-client-request",
+    "x-codex-turn-metadata": json.dumps({"session_id": "planning-fixture-session", "turn_id": "fixture-native-turn"})}), timeout=30) as response:
+    assert json.load(response)["ok"]
+'''
+    binary.write_text(binary.read_text().replace("packet = json.loads(sys.stdin.read()", request + "packet = json.loads(sys.stdin.read()"))
+    planning_env.update(LOOPX_RESEARCH_LOOP="1", LOOPX_ITERATION_CONTEXT="resume", NO_PROXY="127.0.0.1,localhost")
+    with _recording_upstream() as (url, _observed):
+        planning_env["OPENAI_BASE_URL"] = url
+        first = run_once(planning_env)
+        config = (Path(planning_env["LOOPX_CODEX_HOME"]) / "config.toml").read_bytes()
+        second = run_once(planning_env)
+    assert first["ok"] and second["ok"] and second["session"]["action"] == "resume"
+    assert (Path(planning_env["LOOPX_CODEX_HOME"]) / "config.toml").read_bytes() == config
+    assert url.encode() in config
+    for receipt in (first, second):
+        assert receipt["research_loop"]
+        events = [json.loads(line) for line in (Path(planning_env["LOOPX_WAKE_LOG_DIR"]) /
+            receipt["turn_id"] / "provider-requests.jsonl").read_text().splitlines()]
+        finished = next(event for event in events if event["event"] == "finished")
+        assert finished["forwarded"] is True
+        assert finished["wake_id"] == receipt["turn_id"] == finished["loopx_turn_id"]
+        assert finished["native_turn_id"] == "fixture-native-turn"
+
+
 def test_prose_or_fabricated_todo_does_not_qualify_planning(planning_env):
     binary = Path(planning_env["CODEX_BIN"])
     binary.write_text(