fix(benchmark): bind provider admission to each native invocation

benchmark/edgebench/run.py

@@ -123,6 +123,8 @@ def main(argv=None):
                         help="Opt-in short heartbeat context with same-invocation full captures")
     parser.add_argument("--context-delivery", choices=("off", "on"), default="off",
                         help="Experimental native Codex benchmark heartbeat reuse only; off by default")
+    parser.add_argument("--provider-request-admission", action="store_true",
+                        help="Opt-in canonical admission before each heartbeat custom Responses request")
     parser.add_argument("--native-goal-continue", action="store_true",
                         help="Opt-in: when a native Goal declares completion with budget left, start "
                              "continuation Goals until the budget edge (native-goal worker only)")
@@ -197,6 +199,13 @@ def main(argv=None):
         parser.error("--turn-envelope requires a heartbeat worker")
     if args.context_delivery != "off" and not args.worker.startswith("heartbeat-"):
         parser.error("--context-delivery requires a heartbeat worker")
+    if args.provider_request_admission:
+        if not args.worker.startswith("heartbeat-"):
+            parser.error("--provider-request-admission requires a heartbeat worker")
+        if not (api_base_url or "").strip() or not (api_key or "").strip():
+            parser.error("--provider-request-admission requires OPENAI_BASE_URL and OPENAI_API_KEY")
+        if (os.environ.get("CODEX_WIRE_API") or "responses") != "responses":
+            parser.error("--provider-request-admission requires CODEX_WIRE_API=responses")
     if args.native_goal_continue and args.worker != "native-goal":
         parser.error("--native-goal-continue requires --worker native-goal")
     if (args.replan_after_turns is not None or args.replan_after_todos is not None
@@ -345,6 +354,7 @@ def main(argv=None):
                          task_entry=args.task_entry,
                          turn_envelope=args.turn_envelope,
                          context_delivery=args.context_delivery,
+                         **({"provider_request_admission": True} if args.provider_request_admission else {}),
                          native_goal_continue=args.native_goal_continue,
                          replan_after_turns=args.replan_after_turns,
                          replan_after_todos=args.replan_after_todos,
@@ -412,6 +422,7 @@ def main(argv=None):
         "sforge_source": active_source_receipt(),
         **turn_envelope_receipt(explicit=args.turn_envelope),
         **({"context_delivery": "on"} if args.context_delivery != "off" else {}),
+        **({"provider_request_admission": "on"} if args.provider_request_admission else {}),
         **({"native_goal_continue": True} if args.native_goal_continue else {}),
         "task_sha256": hashlib.sha256(task_file.read_bytes()).hexdigest(),
         "feedback": args.feedback, "internet": task.internet,

benchmark/runtime/RUNTIME.md

@@ -417,6 +417,48 @@ approval in its envelope; publishing and production actions keep their gates.
 Staging uses Harbor upload/exec methods, without Docker-label container discovery.
 Backend-specific networking stays in the benchmark's native launcher.
 
+### Opt-in provider request admission
+
+Set `provider_request_admission: true` in the Harbor heartbeat agent's `kwargs`,
+or pass `--provider-request-admission` to an EdgeBench heartbeat worker, to check
+the existing canonical admission before each model-provider request. This is a
+separate boolean opt-in from context delivery and turn envelopes. It requires a
+custom `OPENAI_BASE_URL`, `OPENAI_API_KEY` and the `responses` wire API; login-file
+routes and other execution modes reject activation before execution. The worker
+receives `LOOPX_PROVIDER_REQUEST_ADMISSION=on`, and enabled run/profile receipts
+record `provider_request_admission: "on"`. Omission keeps the existing environment
+and receipt shape. Disable it by omitting the option on a new trial.
+
+Each native invocation gets its own listener and random request path, bound once
+to that invocation. A successor never rebinds the earlier listener to a new
+process or admission identity. The request check reuses the existing Stop, quota
+and canonical admission owners; it does not grant additional work or invent a
+second quota counter. Inbound parsing, upstream connection setup and waiting for
+the response do not hold the Goal/registry lock. The admitted POST dispatch has
+an absolute two-second deadline, shortened by a smaller remaining phase budget.
+An uncertain or partially sent request remains unknown and is not retried by
+this transport.
+
+Per-invocation receipts distinguish requests started, bytes fully sent, pending
+responses and unknown partial sends. `sent` does not certify provider acceptance
+or completion; a response still pending when the worker exits remains visible
+as pending. Request observations contain identities and transport outcomes,
+without request/response bodies, query strings or credentials. Their file writes
+run outside the invocation state lock and the canonical admission scope.
+
+The isolated Codex home and checkpoint configuration retain the existing
+provider route. Only the native process command receives a temporary route
+override for its invocation. SForge may already have selected a host credential
+gateway, including an HTTP route; admission forwards to that same effective
+route read back from the trial, with its existing credential contract. A loopback
+listener alone is not a credential or OS isolation boundary. The caller still
+owns provider access, credential isolation and the worker's process/filesystem
+boundary. This option changes request admission, not provider identity, scoring
+or benchmark qualification.
+For HTTPS upstreams, an explicit proxy uses HTTP CONNECT; unsupported proxy
+layouts refuse activation before launch. A slow upload can reach the bounded
+send deadline and remain unknown even when the upstream has received some bytes.
+
 ## Results, timeout and recovery
 
 ### Live session observation

benchmark/runtime/checkpoint_wake.py

@@ -210,7 +210,7 @@ def _write_input(self, process, body, command, deadline, timeout, *, phase_limit
             stream.close()  # No buffered writes: all input used the raw pipe.
 
     def run(self, execution, body, wake, session_wake, stdout, stderr, timeout, *, phase_limited=False):
-        from .worker import child_process, heartbeat_launch_check, native_command
+        from .worker import native_child_process, native_command
         from .sessions import BenchmarkSessionWake
 
         deadline = time.monotonic() + timeout if timeout is not None else float("inf")
@@ -223,10 +223,10 @@ def run(self, execution, body, wake, session_wake, stdout, stderr, timeout, *, p
                 raise subprocess.TimeoutExpired("checkpoint source wake", timeout)
             command = native_command(launch_env, execution, "execute", wake, session_wake)
             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:
+            with native_child_process(command, env=launch_env, stdout=stdout, stderr=stderr,
+                    session_wake=session_wake, receipt=self.receipt, wake=wake,
+                    timeout=remaining if timeout is not None else None,
+                    original_quota=original_quota) as process:
                 self.receipt["host_invoked"] = True
                 self._write_input(process, body, command, deadline, timeout,
                                   phase_limited=phase_limited)

benchmark/runtime/harbor.py

@@ -75,6 +75,7 @@ def __init__(
         planning_timeout_sec=None,
         turn_envelope=False,
         context_delivery="off",
+        provider_request_admission=False,
         native_goal_continue=False,
         task_objective=None,
         execution_observation=None,
@@ -101,6 +102,11 @@ def __init__(
             turn_envelope,
             context_delivery,
         )
+        if type(provider_request_admission) is not bool:
+            raise ValueError("provider_request_admission must be a boolean opt-in")
+        if provider_request_admission and self.execution.mode != "heartbeat":
+            raise ValueError("provider_request_admission requires heartbeat execution")
+        self.provider_request_admission = provider_request_admission
         if planning_timeout_sec is not None:
             raise ValueError(
                 "Remove planning_timeout_sec; planning shares scheduler_timeout_sec with execution"
@@ -173,6 +179,13 @@ def __init__(
             raise ValueError("Task objective must be a non-empty string")
         self.task_objective = task_objective
         super().__init__(*args, **kwargs)
+        if self.provider_request_admission:
+            if not (self._get_env("OPENAI_BASE_URL") or "").strip() or not (
+                self._get_env("OPENAI_API_KEY") or ""
+            ).strip():
+                raise ValueError("provider_request_admission requires a custom provider URL and API key")
+            if (self._get_env("CODEX_WIRE_API") or "responses") != "responses":
+                raise ValueError("provider_request_admission requires the Responses wire API")
 
     def _turn_envelope_receipt(self) -> dict:
         return {**turn_envelope_receipt(explicit=self.execution.turn_envelope),
@@ -369,6 +382,7 @@ async def install(self, environment: BaseEnvironment) -> None:
                if self.execution_observation is not None else {}),
             **self._turn_envelope_receipt(),
             **self._planning_receipt(),
+            **({"provider_request_admission": "on"} if self.provider_request_admission else {}),
             "home_scope": "trial",
             "loopx_usage_ping_enabled": False,
             "login_shell_node_path": _BASH_ENV,
@@ -621,6 +635,7 @@ def _worker_env(self, *, cwd: str) -> dict[str, str]:
                    if self.checkpoint_request_path is not None else {}),
                 **({"LOOPX_TURN_ENVELOPE": "1"} if self.execution.turn_envelope else {}),
                 **({"LOOPX_CONTEXT_DELIVERY": "on"} if self.execution.context_delivery != "off" else {}),
+                **({"LOOPX_PROVIDER_REQUEST_ADMISSION": "on"} if self.provider_request_admission else {}),
                 **({"LOOPX_NATIVE_GOAL_CONTINUE": "1"} if self.native_goal_continue else {}),
                 "LOOPX_ITERATION_CONTEXT": self.execution.context,
                 "LOOPX_CODEX_SANDBOX": self.execution.sandbox,
@@ -750,6 +765,7 @@ def token_delta(key: str) -> int | None:
             "task_entry": self.execution.task_entry,
             **self._turn_envelope_receipt(),
             **self._planning_receipt(),
+            **({"provider_request_admission": "on"} if self.provider_request_admission else {}),
             "home_scope": "trial",
             **self._replan_receipt(),
             "benchmark_phase": self._phase_number,