diff --git a/benchmark/edgebench/run.py b/benchmark/edgebench/run.py index 36984524f0..98f65a8f5e 100644 --- a/benchmark/edgebench/run.py +++ b/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, diff --git a/benchmark/runtime/RUNTIME.md b/benchmark/runtime/RUNTIME.md index 9e2534610f..bc81447238 100644 --- a/benchmark/runtime/RUNTIME.md +++ b/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 diff --git a/benchmark/runtime/checkpoint_wake.py b/benchmark/runtime/checkpoint_wake.py index 3c2f4bc5ee..6ba8788b6c 100644 --- a/benchmark/runtime/checkpoint_wake.py +++ b/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) diff --git a/benchmark/runtime/harbor.py b/benchmark/runtime/harbor.py index e1a3e2ada5..8733c3873d 100644 --- a/benchmark/runtime/harbor.py +++ b/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, diff --git a/benchmark/runtime/provider_request_admission.py b/benchmark/runtime/provider_request_admission.py new file mode 100644 index 0000000000..41d3757757 --- /dev/null +++ b/benchmark/runtime/provider_request_admission.py @@ -0,0 +1,213 @@ +"""Adapt one native invocation to the existing Goal/Turn admission owner. + +This is transport, not a second quota policy. An invocation never rebinds its +authority, process or gateway; checkpoint continuations create new instances. +""" +from __future__ import annotations + +import copy +import json +import math +import os +import signal +import threading +import time +import uuid +from contextlib import contextmanager +from dataclasses import dataclass +from types import SimpleNamespace +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 + + +def request_admission_enabled(env, mode): + value = env.get("LOOPX_PROVIDER_REQUEST_ADMISSION", "off") + if value not in {"off", "on"}: + raise ValueError("LOOPX_PROVIDER_REQUEST_ADMISSION supports only off or on") + if value == "off": + return False + if (mode != "heartbeat" or not env.get("OPENAI_BASE_URL") or not env.get("OPENAI_API_KEY") + or env.get("CODEX_WIRE_API", "responses") != "responses"): + raise ValueError("provider request admission requires a custom Responses heartbeat provider") + return True + + +@dataclass(frozen=True) +class _Invocation: + admission: object + turn_id: str + session_id: str | None + deadline: float + + +class ProviderRequestAdmission: + """One immutable owner plus a one-shot process attachment.""" + + def __init__(self, env, session_wake, receipt, wake, *, timeout=None, original_quota=None): + from .worker import PHASE_WORK_RESERVE_SECONDS, heartbeat_launch_check + + if session_wake is None or session_wake.admission is None: + raise ValueError("provider request admission requires the original heartbeat session") + captured_env = dict(env) + captured_session = SimpleNamespace(goal_ref=copy.deepcopy(session_wake.goal_ref)) + check = heartbeat_launch_check(captured_env, captured_session, + original_quota=copy.deepcopy(original_quota)) + deadline = math.inf + if timeout is not None: + if not math.isfinite(timeout) or timeout < 0: + raise ValueError("provider request timeout must be finite and non-negative") + deadline = time.monotonic() + timeout + if captured_env.get("LOOPX_PHASE_DEADLINE_EPOCH"): + epoch = float(captured_env["LOOPX_PHASE_DEADLINE_EPOCH"]) + if not math.isfinite(epoch): + raise ValueError("provider phase deadline must be finite") + deadline = min(deadline, + time.monotonic() + epoch - PHASE_WORK_RESERVE_SECONDS - time.time()) + self._invocation = _Invocation( + session_wake.admission.with_launch_decision(check), captured_env["LOOPX_TURN"], + session_wake.session_id, deadline) + self._env = captured_env + self._process = None + self._retired = False + self._ready = threading.Event() + self._lock = threading.RLock() + self._log_lock = threading.Lock() + self._request_events = {} + self.refusal_code = None + self._wake = wake + self._summary = { + "invocation_id": uuid.uuid4().hex, "turn_id": self._invocation.turn_id, + "session_id": self._invocation.session_id, "requests": 0, + "pending": 0, "sent": 0, "unknown": 0, "refused": 0, + } + receipt.setdefault("provider_request_invocations", []).append(self._summary) + + def started(self, process): + with self._lock: + if self._process is not None or self._retired: + raise ValueError("provider request invocation cannot be rebound") + self._process = process + self._ready.set() + + def retire(self): + with self._lock: + self._retired = True + self._ready.set() + + def before_cleanup(self, _process, _scope_error): + self.retire() + + def signal_requested(self, _signal): + pass + + def after_cleanup(self, _process, _cleanup_error): + pass + + def _reject(self, code): + with self._lock: + # A delayed request from a retired invocation cannot poison the + # receipt or process of a checkpoint replacement. + if not self._retired: + self.refusal_code = self.refusal_code or code + self._summary["refusal_code"] = self.refusal_code + if self._process is not None and self._process.poll() is None: + try: + os.killpg(self._process.pid, signal.SIGINT) + except ProcessLookupError: + pass + raise ProviderGatewayRequestRejected(code) + + def _require_live(self): + if self.refusal_code is not None: + self._reject(self.refusal_code) + if self._retired or self._process is None or self._process.poll() is not None: + self._reject("provider_request_without_live_owner") + if time.monotonic() >= self._invocation.deadline: + self._reject("provider_request_budget_exhausted") + + @contextmanager + def lifetime(self): + # Popen may start a request before the parent receives its handle. Wait + # only for this invocation's single attachment, never for a successor. + self._ready.wait(timeout=max(0, min(2, self._invocation.deadline - time.monotonic()))) + with self._lock: + self._require_live() + entered = False + try: + with self._invocation.admission.launch_lifetime(operation="provider_request_dispatch"): + entered = True + with self._lock: + # Authority/registry contention can outlive the process or + # phase. Revalidate after every wait and before any POST. + self._require_live() + yield min(time.monotonic() + 2, self._invocation.deadline) + except ProviderGatewayRequestRejected: + raise + except FirstPartyHostRuntimeRejected as error: + self._reject(error.code) + except Exception: + if entered: + raise # A transport failure is not a quota refusal or a retry grant. + self._reject("provider_request_admission_unavailable") + + def observe(self, request): + with self._lock: + request_id, phase = request["request_id"], request["event"] + prior = self._request_events.get(request_id) + if phase == "started" and prior is None: + self._summary["requests"] += 1 + self._summary["pending"] += 1 + elif phase == "dispatch" and prior == "started": + self._summary["sent"] += 1 + elif phase == "finished" and prior in {"started", "dispatch"}: + self._summary["pending"] -= 1 + if request.get("forwarded") is None: + self._summary["unknown"] += 1 + if "refusal_code" in request: + self._summary["refused"] += 1 + else: + return # Re-observing a phase cannot count one request twice. + self._request_events[request_id] = phase + event = {"invocation_id": self._summary["invocation_id"], + "turn_id": self._invocation.turn_id, "session_id": self._invocation.session_id, + **request} + # A slow filesystem must not hold the invocation state lock, which a + # later request also needs inside its canonical admission lifetime. + with self._log_lock: + with (self._wake / "provider-requests.jsonl").open("a", encoding="utf-8") as stream: + stream.write(json.dumps(event, sort_keys=True) + "\n") + + @contextmanager + def serve(self, command): + upstream = urlsplit(self._env["OPENAI_BASE_URL"]) + proxy = None + bypass = self._env.get("no_proxy", self._env.get("NO_PROXY", "")) + if upstream.scheme == "https" and not proxy_bypass_environment(upstream.netloc, {"no": bypass}): + proxy = self._env.get("https_proxy", self._env.get("HTTPS_PROXY")) or None + try: + with serve_runner_owned_provider_gateway( + upstream_base_url=self._env["OPENAI_BASE_URL"], + upstream_bearer_token=self._env["OPENAI_API_KEY"], + request_lifetime=self.lifetime, observe_request=self.observe, + upstream_proxy_url=proxy, + # SForge can supply a private host credential gateway over + # HTTP. Preserve the runner's effective route; do not move + # credentials or silently replace it with the public URL. + allow_http_upstream=True) as gateway: + self.base_url = gateway.base_url + # Keep the durable home on its actual provider for checkpoint + # compatibility; only this native invocation gets the gateway. + routed = list(command) + routed[1:1] = ["-c", "model_providers.harbor.base_url=" + json.dumps(gateway.base_url)] + try: + yield routed + finally: + self.retire() + finally: + self.retire() diff --git a/benchmark/runtime/sforge.py b/benchmark/runtime/sforge.py index 1e9abdbcab..3ccb5e10cc 100644 --- a/benchmark/runtime/sforge.py +++ b/benchmark/runtime/sforge.py @@ -94,6 +94,7 @@ def __init__(self, config, *, profile: str, cwd: str, task_entry: str | None = None, turn_envelope: bool = False, context_delivery: str = "off", + provider_request_admission: bool = False, native_goal_continue: bool = False, replan_after_turns: int | None = None, replan_after_todos: int | None = None, @@ -114,6 +115,16 @@ def __init__(self, config, *, profile: str, cwd: str, mode="heartbeat" if profile.startswith("heartbeat-") else "plain", context_delivery=context_delivery, ).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 not profile.startswith("heartbeat-"): + raise ValueError("provider_request_admission requires a heartbeat profile") + if provider_request_admission: + if not (config.agent_api_base_url or "").strip() or not (config.agent_api_key or "").strip(): + raise ValueError("provider_request_admission requires a custom provider URL and API key") + if (os.environ.get("CODEX_WIRE_API") or "responses") != "responses": + raise ValueError("provider_request_admission requires the Responses wire API") + self.provider_request_admission = provider_request_admission if timeout_seconds <= PHASE_WORK_RESERVE_SECONDS: raise ValueError("Worker budget must exceed the 160s startup/settlement reserve") if not config.agent_model or not config.agent_effort: @@ -304,6 +315,7 @@ def _install_worker(self, backend, handle, log_dir, logger): task_entry=self.task_entry, turn_envelope=self.turn_envelope, context_delivery=self.context_delivery, + **({"provider_request_admission": True} if self.provider_request_admission else {}), native_goal_continue=self.native_goal_continue, replan_after_turns=self.replan_after_turns, replan_after_todos=self.replan_after_todos, @@ -392,6 +404,7 @@ def _install_worker(self, backend, handle, log_dir, logger): if self.task_entry == "loopx-planned" else {}), "stop_hook": self.profile == "official", **(self.runtime._turn_envelope_receipt() if self.runtime is not None else {}), + **({"provider_request_admission": "on"} if self.provider_request_admission else {}), **({"native_goal_continue": True} if self.native_goal_continue else {}), "outer_resume": self.resume_cmd is not None, "explore_graph": self.profile == "heartbeat-explore", diff --git a/benchmark/runtime/worker.py b/benchmark/runtime/worker.py index 9479fe3e27..6603790d58 100644 --- a/benchmark/runtime/worker.py +++ b/benchmark/runtime/worker.py @@ -194,6 +194,30 @@ def check(registry_path): return check +@contextmanager +def native_child_process(command, *, env, stdout, stderr, session_wake, receipt, wake, + timeout=None, original_quota=None): + """Share invocation ownership across ordinary and checkpoint launches.""" + guard = None + scope = nullcontext(command) + if env.get("LOOPX_PROVIDER_REQUEST_ADMISSION", "off") == "on": + from .provider_request_admission import ProviderRequestAdmission + + guard = ProviderRequestAdmission(env, session_wake, receipt, wake, + timeout=timeout, original_quota=original_quota) + scope = guard.serve(command) + with scope as routed: + with child_process(routed, 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, original_quota=original_quota), + **({"process_observation": guard} if guard is not None else {})) as process: + yield process + if guard is not None and guard.refusal_code is not None: + from loopx.control_plane.goals.first_party_host_admission import FirstPartyHostRuntimeRejected + + raise FirstPartyHostRuntimeRejected(guard.refusal_code) + + def _reap_child_process(process, *, process_observation=None): """Reap the process group created by child_process, including descendants.""" try: @@ -468,6 +492,11 @@ def run_once(env: dict[str, str]) -> dict: turn_envelope=env.get("LOOPX_TURN_ENVELOPE", "0") == "1", context_delivery=env.get("LOOPX_CONTEXT_DELIVERY", "off"), ) + request_admission = False + if env.get("LOOPX_PROVIDER_REQUEST_ADMISSION", "off") != "off": + from .provider_request_admission import request_admission_enabled + + request_admission = request_admission_enabled(env, execution.mode) stage = env.get("LOOPX_TASK_STAGE", "execute") if stage not in {"plan", "execute"} or (stage == "plan" and execution.task_entry != "loopx-planned"): raise ValueError("invalid task-entry stage") @@ -485,6 +514,11 @@ def run_once(env: dict[str, str]) -> dict: # Preserve this trial's resolved off/on choice in both preflight and solver # subprocesses; an omitted off must not inherit a machine-wide opt-in. env["LOOPX_CONTEXT_DELIVERY"] = execution.context_delivery + # The resolved trial treatment must not inherit an ambient parent opt-in. + if request_admission: + env["LOOPX_PROVIDER_REQUEST_ADMISSION"] = "on" + else: + env.pop("LOOPX_PROVIDER_REQUEST_ADMISSION", None) if execution.context_delivery == "on": # The CLI validates this declaration before enabling native transport. # Carry the resolved default as well as an explicitly supplied mode. @@ -503,6 +537,7 @@ def run_once(env: dict[str, str]) -> dict: **({"turn_envelope": True, "turn_envelope_source": "explicit"} if execution.turn_envelope else {}), **({"context_delivery": "on"} if execution.context_delivery != "off" else {}), + **({"provider_request_admission": "on"} if request_admission else {}), "stage": stage, "home_scope": "trial", "ok": False, @@ -644,9 +679,9 @@ def run_once(env: dict[str, str]) -> dict: phase_limited=phase_limited) receipt["ok"] = receipt["return_code"] == 0 else: - 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: + with native_child_process(command, env=env, stdout=stdout, stderr=stderr, + session_wake=session_wake, receipt=receipt, wake=wake, + timeout=timeout) as process: if session_wake is not None: receipt["host_invoked"] = True timeout, phase_limited = _phase_timeout(timeout, phase_work_deadline) diff --git a/benchmark/tests/test_provider_request_admission.py b/benchmark/tests/test_provider_request_admission.py new file mode 100644 index 0000000000..ceaa80b4d9 --- /dev/null +++ b/benchmark/tests/test_provider_request_admission.py @@ -0,0 +1,466 @@ +"""Invocation ownership across real HTTP waits and canonical Goal decisions. + +Only transport scheduling is instrumented. Admission, registry locks, File / +SQLite stores, process liveness and provider bytes use their actual owners. +These synthetic providers do not qualify any external model or benchmark score. +""" +from __future__ import annotations + +from concurrent.futures import Future +from contextlib import contextmanager +from dataclasses import dataclass +import http.client +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import json +from pathlib import Path +import socket +import subprocess +import sys +import threading +import time +from types import SimpleNamespace +from urllib.parse import urlsplit + +import pytest + +from benchmark.runtime.provider_request_admission import ( + ProviderRequestAdmission, + request_admission_enabled, +) +from loopx.capabilities.benchmark_toolkit import provider_gateway +from loopx.control_plane.goals import activation_service, first_party_host_admission +from loopx.control_plane.goals.first_party_host_admission import FirstPartyHostGoalAdmission +from loopx.control_plane.projects.registry_codec import project_registry_transaction +from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime +from tests.control_plane import test_quota_settlement_cli as quota_fixture + + +@dataclass +class _Authority: + project: Path + runtime: Path + registry: Path + root: Path + api_key: str = "fixture-provider-key" + + def invocation(self, upstream, name, *, timeout=None, receipt=None, wake=None, turn_id=None): + if wake is None: + wake = self.root / name + wake.mkdir() + env = { + "LOOPX_PROJECT": str(self.project), + "LOOPX_RUNTIME_ROOT": str(self.runtime), + "LOOPX_GOAL_ID": quota_fixture.GOAL_ID, + "LOOPX_AGENT_ID": quota_fixture.AGENT_ID, + "LOOPX_TURN": turn_id or name, + "OPENAI_BASE_URL": upstream, + "OPENAI_API_KEY": self.api_key, + } + session = SimpleNamespace( + admission=FirstPartyHostGoalAdmission.for_plan( + registry_path=self.registry, goal_id=quota_fixture.GOAL_ID, + planned_goal_ref=None, + ), + goal_ref=None, session_id=f"session-{name}", + ) + if receipt is None: + receipt = {} + guard = ProviderRequestAdmission(env, session, receipt, wake, timeout=timeout) + return guard, receipt, wake + + def stop(self): + result = activation_service.set_goal_activation_state( + registry_path=self.registry, goal_id=quota_fixture.GOAL_ID, + state="stopped", reason="Stop this synthetic provider invocation.", + runtime_root_override=str(self.runtime), actor_kind="owner", execute=True, + ) + assert result["written"], result + return result + + +@pytest.fixture(params=["file", "sqlite"]) +def authority(tmp_path, monkeypatch, request): + isolate_sqlite_runtime(tmp_path, monkeypatch) + monkeypatch.setenv("LOOPX_USAGE_PING", "0") + project, runtime, registry = quota_fixture._write_fixture( + tmp_path, provider=request.param, + ) + code, current = quota_fixture._run_cli( + registry, runtime, "quota", "should-run", "--goal-id", quota_fixture.GOAL_ID, + "--agent-id", quota_fixture.AGENT_ID, "--runtime-profile", "generic_cli", + ) + assert code == 0 and current["should_run"] is True, current + return _Authority(project, runtime, registry, tmp_path) + + +@contextmanager +def _process(): + process = subprocess.Popen( + [sys.executable, "-c", "import time; time.sleep(60)"], + start_new_session=True, stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + ) + try: + yield process + finally: + if process.poll() is None: + process.terminate() + try: + process.wait(timeout=5) + except subprocess.TimeoutExpired: + process.kill() + process.wait(timeout=5) + + +def _async(function): + result = Future() + + def run(): + try: + result.set_result(function()) + except BaseException as error: + result.set_exception(error) + + threading.Thread(target=run, daemon=True).start() + return result + + +@contextmanager +def _upstream(*, hold_body=False, hold_response=False): + state = SimpleNamespace( + rows=[], headers=threading.Event(), body=threading.Event(), + read_body=threading.Event(), respond=threading.Event(), + ) + if not hold_body: + state.read_body.set() + if not hold_response: + state.respond.set() + + class Handler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.0" + + def setup(self): + # Small receive windows plus a multi-megabyte body establish real + # backpressure; a 128 KiB write can fit entirely in kernel buffers. + self.request.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 16 * 1024) + self.request.settimeout(5) + super().setup() + + def log_message(self, _format, *args): + pass + + def do_POST(self): + row = {"path": self.path, "authorization": self.headers.get("Authorization"), + "length": int(self.headers["Content-Length"])} + state.rows.append(row) + state.headers.set() + try: + assert state.read_body.wait(12), "test did not release the upstream body reader" + row["body"] = self.rfile.read(row["length"]) + state.body.set() + assert state.respond.wait(12), "test did not release the upstream response" + body = b'{"id":"synthetic-response"}' + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + except (OSError, AssertionError) as error: + row["transport_error"] = type(error).__name__ + state.body.set() + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + server.daemon_threads = True + state.base_url = f"http://127.0.0.1:{server.server_port}/v1" + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield state + finally: + state.read_body.set() + state.respond.set() + server.shutdown() + server.server_close() + thread.join(timeout=5) + + +def _post(base_url, body=b"{}", *, path="/responses"): + parsed = urlsplit(base_url) + connection = http.client.HTTPConnection(parsed.hostname, parsed.port, timeout=8) + try: + connection.request("POST", parsed.path + path, body=body, + headers={"Content-Type": "application/json"}) + response = connection.getresponse() + return response.status, json.loads(response.read()) + finally: + connection.close() + + +def _finished(wake, *, count=1, invocation_id=None): + # A framed HTTP error body can reach the caller before the handler's + # finally block. Await the observation itself, not an incidental socket EOF. + deadline = time.monotonic() + 3 + while True: + path = wake / "provider-requests.jsonl" + rows = [json.loads(line) for line in path.read_text().splitlines()] if path.exists() else [] + finished = [row for row in rows if row.get("event") == "finished" + and (invocation_id is None or row["invocation_id"] == invocation_id)] + if len(finished) >= count: + return finished + assert time.monotonic() < deadline, "request did not publish its finished observation" + threading.Event().wait(0.01) + + +@pytest.mark.parametrize("decision", ["allowed", "stop", "quota_wait"]) +def test_each_request_rechecks_canonical_authority(authority, decision): + with _upstream() as upstream, _process() as process: + guard, receipt, wake = authority.invocation(upstream.base_url, "live") + with guard.serve(["native", "-"]): + guard.started(process) + assert _post(guard.base_url)[0] == 200 + if decision == "stop": + authority.stop() + elif decision == "quota_wait": + source_before = authority.registry.read_bytes() + code, spent = quota_fixture._run_cli( + authority.registry, authority.runtime, "quota", "spend-slot", + "--goal-id", quota_fixture.GOAL_ID, "--agent-id", quota_fixture.AGENT_ID, + "--source", "controller", "--slots", "2", "--execute", + ) + assert code == 0 and spent["appended"], spent + assert spent["before"]["should_run"] is True + assert spent["after"]["should_run"] is False + # Exercise the live quota reader, not registry witness mismatch. + assert authority.registry.read_bytes() == source_before + status, payload = _post(guard.base_url) + if decision == "allowed": + assert status == 200 and process.poll() is None + assert len(upstream.rows) == 2 + else: + assert status == 409, payload + assert payload["error"]["code"] == ( + "goal_stopped" if decision == "stop" else "goal_launch_not_allowed" + ) + process.wait(timeout=5) + assert len(upstream.rows) == 1, "a refused second request reached the provider" + assert all(row["body"] == b"{}" for row in upstream.rows) + assert all(row["authorization"] == f"Bearer {authority.api_key}" for row in upstream.rows) + assert len(_finished(wake, count=2)) == 2 + summary = receipt["provider_request_invocations"][0] + assert summary["requests"] == 2 and summary["sent"] == len(upstream.rows) + + +@pytest.mark.parametrize("expired", ["process", "deadline"]) +def test_registry_wait_rechecks_original_process_and_deadline(authority, monkeypatch, expired): + waiting = threading.Event() + original = first_party_host_admission.project_registry_transaction + + @contextmanager + def observed_transaction(*args, **kwargs): + if kwargs.get("operation") == "provider_request_dispatch": + waiting.set() + with original(*args, **kwargs) as transaction: + yield transaction + + monkeypatch.setattr(first_party_host_admission, "project_registry_transaction", observed_transaction) + with _upstream() as upstream, _process() as process: + guard, _receipt, wake = authority.invocation( + upstream.base_url, "contended", timeout=1.5 if expired == "deadline" else None, + ) + with guard.serve(["native", "-"]): + guard.started(process) + # This is the real writer lock, not a substituted admission result. + with project_registry_transaction(authority.registry, operation="test_writer_holds_registry"): + request = _async(lambda: _post(guard.base_url)) + assert waiting.wait(3), "request did not reach the held registry lock" + assert not request.done() and not upstream.rows + if expired == "process": + process.terminate() + process.wait(timeout=5) + else: + # A real deadline elapses while the actual authority lock is + # unavailable. No clock or live-owner predicate is replaced. + threading.Event().wait(1.6) + status, payload = request.result(timeout=6) + assert status == 409, payload + assert payload["error"]["code"] == ( + "provider_request_without_live_owner" if expired == "process" + else "provider_request_budget_exhausted" + ) + assert not upstream.rows and _finished(wake)[0]["forwarded"] is False + + +def test_process_attachment_is_one_shot(authority): + with _upstream() as upstream, _process() as old, _process() as replacement: + guard, _receipt, _wake = authority.invocation(upstream.base_url, "one-shot") + guard.started(old) + with pytest.raises(ValueError, match="cannot be rebound"): + guard.started(replacement) + guard.retire() + with pytest.raises(ValueError, match="cannot be rebound"): + guard.started(replacement) + assert replacement.poll() is None + + +@pytest.mark.parametrize("stage,shared_receipt", [ + ("backlog", False), ("accepted", False), ("headers", False), ("body", False), + ("upstream_connect", False), ("upstream_connect", True), +]) +def test_delayed_old_request_cannot_adopt_replacement(authority, monkeypatch, stage, shared_receipt): + waiting, release = threading.Event(), threading.Event() + old_port = [None] + real_accept = provider_gateway.ThreadingHTTPServer.get_request + real_connect = http.client.HTTPConnection.connect + connect_claimed = threading.Event() + upstream_port = [None] + + def observed_accept(server): + if server.server_port == old_port[0] and stage == "backlog": + waiting.set() + assert release.wait(10), "test did not release the actual listen backlog" + result = real_accept(server) + if server.server_port == old_port[0] and stage == "accepted": + waiting.set() + assert release.wait(10), "test did not release the accepted old socket" + if server.server_port == old_port[0] and stage in {"headers", "body"}: + waiting.set() + return result + + def observed_connect(connection): + real_connect(connection) + if (stage == "upstream_connect" and connection.port == upstream_port[0] + and not connect_claimed.is_set()): + connect_claimed.set() + waiting.set() + assert release.wait(10), "test did not release upstream connection preparation" + + monkeypatch.setattr(provider_gateway.ThreadingHTTPServer, "get_request", observed_accept) + monkeypatch.setattr(http.client.HTTPConnection, "connect", observed_connect) + raw = None + with _upstream() as upstream, _process() as old, _process() as replacement: + upstream_port[0] = urlsplit(upstream.base_url).port + original, original_receipt, original_wake = authority.invocation(upstream.base_url, "old") + try: + with original.serve(["native", "-"]): + original.started(old) + parsed = urlsplit(original.base_url) + old_port[0] = parsed.port + if stage == "upstream_connect": + pending = _async(lambda: _post(original.base_url)) + else: + raw = socket.create_connection((parsed.hostname, parsed.port), timeout=5) + header = (f"POST {parsed.path}/responses HTTP/1.0\r\n" + "Host: localhost\r\nContent-Length: 2\r\n").encode() + raw.sendall(header + ( + b"\r\n{}" if stage in {"backlog", "accepted"} + else b"\r\n{" if stage == "body" else b"" + )) + assert waiting.wait(3), f"old request did not reach {stage}" + assert not upstream.rows + original.retire() + old.terminate() + old.wait(timeout=5) + shared = {"receipt": original_receipt, "wake": original_wake, "turn_id": "old"} if shared_receipt else {} + successor, successor_receipt, successor_wake = authority.invocation(upstream.base_url, "new", **shared) + with successor.serve(["native", "-"]): + successor.started(replacement) + assert _post(successor.base_url)[0] == 200 + release.set() + if stage == "headers": + raw.sendall(b"\r\n{}") + elif stage == "body": + raw.sendall(b"}") + if raw is not None: + response = http.client.HTTPResponse(raw) + response.begin() + status, payload = response.status, json.loads(response.read()) + else: + status, payload = pending.result(timeout=6) + assert status == 409 and payload["error"]["code"] == "provider_request_without_live_owner" + assert original.refusal_code is None, "a retired request latched a later invocation refusal" + assert successor.refusal_code is None and replacement.poll() is None + assert len(upstream.rows) == 1 and upstream.rows[0]["body"] == b"{}" + old_summary = original_receipt["provider_request_invocations"][0] + new_summary = successor_receipt["provider_request_invocations"][-1] + old_event, = _finished(original_wake, invocation_id=old_summary["invocation_id"]) + new_event, = _finished(successor_wake, invocation_id=new_summary["invocation_id"]) + assert old_event["turn_id"] == "old" and old_event["forwarded"] is False + assert new_event["turn_id"] == ("old" if shared_receipt else "new") + assert new_event["forwarded"] is True + assert old_event["session_id"] == "session-old" and new_event["session_id"] == "session-new" + assert old_event["invocation_id"] != new_event["invocation_id"] + assert old_summary["sent"] == 0 and old_summary["refused"] == 1 + assert new_summary["sent"] == 1 and new_summary["refused"] == 0 + assert old_summary["requests"] == new_summary["requests"] == 1 + assert old_summary["pending"] == new_summary["pending"] == 0 + assert "refusal_code" not in new_summary + if shared_receipt: + assert original_receipt is successor_receipt and original_wake == successor_wake + assert len(successor_receipt["provider_request_invocations"]) == 2 + # Address the old path to the new listener, as after port reuse. + new_address = urlsplit(successor.base_url) + assert _post(f"http://{new_address.netloc}{parsed.path}")[0] == 404 + assert replacement.poll() is None and len(upstream.rows) == 1 + finally: + release.set() + if raw is not None: + raw.close() + + +def test_real_backpressure_bounds_writer_wait_and_records_unknown_acceptance(authority): + body = b"x" * (8 * 1024 * 1024) + with _upstream(hold_body=True) as upstream, _process() as process: + guard, receipt, wake = authority.invocation(upstream.base_url, "backpressure") + with guard.serve(["native", "-"]): + guard.started(process) + request = _async(lambda: _post(guard.base_url, body)) + assert upstream.headers.wait(5), "provider did not receive the POST headers" + assert upstream.rows[0]["length"] == len(body) + stopped = _async(authority.stop) + try: + # The provider still has not read its body. A canonical Stop + # writer must complete once the bounded partial dispatch ends. + assert stopped.result(timeout=6)["written"] + assert not upstream.read_body.is_set() + status, payload = request.result(timeout=3) + assert status == 502, payload + event = _finished(wake)[0] + assert event["forwarded"] is None and "refusal_code" not in event + assert guard.refusal_code is None and process.poll() is None + summary = receipt["provider_request_invocations"][0] + assert summary["unknown"] == 1 and summary["sent"] == 0 + assert summary["refused"] == 0 + finally: + upstream.read_body.set() + assert upstream.body.wait(5) + assert 0 < len(upstream.rows[0]["body"]) < len(body), "fixture did not establish a partial send" + assert _post(guard.base_url)[0] == 409 + assert len(upstream.rows) == 1, "unknown acceptance must not retry the POST" + + +def test_slow_response_does_not_hold_goal_writer(authority): + with _upstream(hold_response=True) as upstream, _process() as process: + guard, receipt, wake = authority.invocation(upstream.base_url, "slow-response") + with guard.serve(["native", "-"]): + guard.started(process) + request = _async(lambda: _post(guard.base_url)) + assert upstream.body.wait(5) + stopped = _async(authority.stop) + try: + assert stopped.result(timeout=4)["written"] + assert not upstream.respond.is_set() and not request.done() + finally: + upstream.respond.set() + assert request.result(timeout=3)[0] == 200 + assert _finished(wake)[0]["forwarded"] is True + assert receipt["provider_request_invocations"][0]["sent"] == 1 + assert _post(guard.base_url)[0] == 409 + assert len(upstream.rows) == 1 + + +@pytest.mark.parametrize("old_flag", [None, "0", "1"]) +def test_existing_research_flag_never_enables_request_admission(old_flag): + env = {} if old_flag is None else {"LOOPX_RESEARCH_LOOP": old_flag} + assert request_admission_enabled(env, "plain") is False + assert request_admission_enabled(env, "heartbeat") is False + assert request_admission_enabled({**env, "LOOPX_PROVIDER_REQUEST_ADMISSION": "off"}, "turn") is False diff --git a/benchmark/tests/test_sforge_runtime.py b/benchmark/tests/test_sforge_runtime.py index db5bf2103a..e9fdaf9bdf 100644 --- a/benchmark/tests/test_sforge_runtime.py +++ b/benchmark/tests/test_sforge_runtime.py @@ -1217,8 +1217,11 @@ def copy(handle, source, target): SForgeWorker(config, profile=profile, cwd='/task') -@pytest.mark.parametrize('profile', ['official', 'native-goal']) -def test_custom_provider_preserves_sforge_runtime_gateway(tmp_path, monkeypatch, profile): +@pytest.mark.parametrize('profile,provider_request_admission', [ + ('official', False), ('native-goal', False), + ('heartbeat-resume', False), ('heartbeat-resume', True), ('heartbeat-explore', True), +]) +def test_custom_provider_preserves_sforge_runtime_gateway(tmp_path, monkeypatch, profile, provider_request_admission): pytest.importorskip('sforge') pytest.importorskip('harbor') import tomllib @@ -1233,6 +1236,8 @@ def test_custom_provider_preserves_sforge_runtime_gateway(tmp_path, monkeypatch, monkeypatch.setenv('CODEX_AUTH_JSON_PATH', str(tmp_path / 'must-not-read-auth.json')) monkeypatch.setenv('OPENAI_API_KEY', provider_key) monkeypatch.setenv('OPENAI_BASE_URL', provider_url) + monkeypatch.setenv('CODEX_WIRE_API', 'responses') + monkeypatch.setenv('LOOPX_PROVIDER_REQUEST_ADMISSION', 'on') async def installed(self, environment): pass @@ -1254,7 +1259,8 @@ def copy(handle, source, target): agent_model='fixture', agent_effort='xhigh', agent_api_base_url=provider_url, agent_api_key=provider_key, ) - worker = SForgeWorker(config, profile=profile, cwd='/task') + worker = SForgeWorker(config, profile=profile, cwd='/task', + provider_request_admission=provider_request_admission) worker.install_stop_hook(backend, None, tmp_path, None) if profile == 'official': @@ -1264,7 +1270,12 @@ def copy(handle, source, target): env = worker.runtime._worker_env(cwd='/task') assert env['OPENAI_BASE_URL'] == runtime_url assert env['OPENAI_API_KEY'] == 'runtime-managed-credential' + assert env.get('LOOPX_PROVIDER_REQUEST_ADMISSION') == ('on' if provider_request_admission else None) worker.prepared = True + assert config.agent_api_base_url == provider_url + receipt = (tmp_path / 'worker-profile.json').read_text() + assert json.loads(receipt).get('provider_request_admission') == ('on' if provider_request_admission else None) + assert provider_key not in receipt command = worker.format_run_cmd('/task/prompt.md') assert provider_url not in command assert provider_key not in command @@ -1272,6 +1283,59 @@ def copy(handle, source, target): assert all(provider_key not in contents for contents in copied.values()) +@pytest.mark.parametrize('option,profile,base_url,api_key,wire_api', [ + ('on', 'heartbeat-resume', 'https://provider.invalid/v1', 'synthetic-key', 'responses'), + (1, 'heartbeat-resume', 'https://provider.invalid/v1', 'synthetic-key', 'responses'), + (None, 'heartbeat-resume', 'https://provider.invalid/v1', 'synthetic-key', 'responses'), + (True, 'official', 'https://provider.invalid/v1', 'synthetic-key', 'responses'), + (True, 'single', 'https://provider.invalid/v1', 'synthetic-key', 'responses'), + (True, 'native-goal', 'https://provider.invalid/v1', 'synthetic-key', 'responses'), + (True, 'heartbeat-resume', None, 'synthetic-key', 'responses'), + (True, 'heartbeat-resume', 'https://provider.invalid/v1', None, 'responses'), + (True, 'heartbeat-resume', 'https://provider.invalid/v1', 'synthetic-key', 'chat'), +]) +def test_sforge_provider_request_admission_rejects_unsupported_activation( + tmp_path, monkeypatch, option, profile, base_url, api_key, wire_api, +): + pytest.importorskip('sforge') + pytest.importorskip('harbor') + from sforge.harness.config import SForgeConfig + from benchmark.runtime.sforge import SForgeWorker + + monkeypatch.setenv('CODEX_AUTH_JSON_PATH', str(tmp_path / 'unused-login.json')) + monkeypatch.setenv('CODEX_WIRE_API', wire_api) + config = SForgeConfig(agent_model='fixture', agent_effort='xhigh', + agent_api_base_url=base_url, agent_api_key=api_key) + with pytest.raises(ValueError, match='provider_request_admission'): + SForgeWorker(config, profile=profile, cwd='/task', provider_request_admission=option) + + +@pytest.mark.parametrize('profile,base_url,wire_api', [ + ('official', 'https://provider.invalid/v1', 'responses'), + ('single', 'https://provider.invalid/v1', 'responses'), + ('native-goal', 'https://provider.invalid/v1', 'responses'), + ('heartbeat-resume', '', 'responses'), + ('heartbeat-resume', 'https://provider.invalid/v1', 'chat'), +]) +def test_edgebench_request_admission_preflight_creates_no_attempt( + tmp_path, monkeypatch, capsys, profile, base_url, wire_api, +): + pytest.importorskip('sforge') + pytest.importorskip('harbor') + from benchmark.edgebench.run import main + + monkeypatch.setenv('OPENAI_BASE_URL', base_url) + monkeypatch.setenv('OPENAI_API_KEY', 'synthetic-provider-key') + monkeypatch.setenv('CODEX_WIRE_API', wire_api) + with pytest.raises(SystemExit, match='2'): + main(['--task', 'fixture', '--tasks-dir', str(tmp_path), '--log-dir', str(tmp_path), + '--run-id', 'unsupported-admission', '--worker', profile, '--model', 'fixture', + '--effort', 'xhigh', '--judge-url', 'http://127.0.0.1:9999', + '--provider-request-admission']) + assert '--provider-request-admission requires' in capsys.readouterr().err + assert not (tmp_path / 'runs').exists() + + def test_custom_provider_without_key_fails_before_creating_attempt(tmp_path, monkeypatch): pytest.importorskip('sforge') pytest.importorskip('harbor') @@ -1285,10 +1349,13 @@ def test_custom_provider_without_key_fails_before_creating_attempt(tmp_path, mon assert not (tmp_path / 'runs').exists() -@pytest.mark.parametrize('custom', [False, True]) +@pytest.mark.parametrize('custom,profile,provider_request_admission', [ + (False, 'official', False), (True, 'official', False), + (True, 'heartbeat-resume', False), (True, 'heartbeat-resume', True), +]) @pytest.mark.parametrize('bridge', [False, True], ids=['docker-default', 'docker-bridge']) -@pytest.mark.parametrize('worker', ['official', 'heartbeat-resume']) -def test_edgebench_cli_merges_sforge_env_with_explicit_runner_config(tmp_path, monkeypatch, custom, bridge, worker): +def test_edgebench_cli_merges_sforge_env_with_explicit_runner_config( + tmp_path, monkeypatch, custom, profile, bridge, provider_request_admission): pytest.importorskip('sforge') pytest.importorskip('harbor') from types import SimpleNamespace @@ -1297,6 +1364,7 @@ def test_edgebench_cli_merges_sforge_env_with_explicit_runner_config(tmp_path, m monkeypatch.setenv('LOOPX_EXPECTED_COMMIT', 'fixture') monkeypatch.setenv('CODEX_AUTH_JSON_PATH', '/synthetic-login') monkeypatch.setenv('OPENAI_API_KEY', 'synthetic-provider-secret') + monkeypatch.setenv('CODEX_WIRE_API', 'responses') attestation_key = tmp_path / 'runtime-attestation.key' attestation_key.write_text('synthetic-attestation-key') monkeypatch.setenv('SFORGE_RUNTIME_ATTESTATION_KEY_FILE', str(attestation_key)) @@ -1327,15 +1395,21 @@ def stop_before_solver(**kwargs): assert config.runtime_attestation_key_file == attestation_key assert config.runner_source_revision == 'a' * 40 assert config.work_cpu_limit == 4 + assert kwargs['agent'].provider_request_admission is provider_request_admission raise RuntimeError('synthetic launch boundary') monkeypatch.setattr(run, 'run_agent', stop_before_solver) + args = ['--task', 'fixture', '--tasks-dir', str(tmp_path), '--log-dir', str(tmp_path), + '--run-id', 'route', '--worker', profile, '--model', 'fixture', + '--effort', 'xhigh', '--judge-url', 'http://127.0.0.1:9999'] + if provider_request_admission: + args += ['--provider-request-admission'] + if bridge: + args += ['--docker-host-bridge'] with pytest.raises(RuntimeError, match='synthetic launch boundary'): - run.main(['--task', 'fixture', '--tasks-dir', str(tmp_path), '--log-dir', str(tmp_path), - '--run-id', 'route', '--worker', worker, '--model', 'fixture', - '--effort', 'xhigh', '--judge-url', 'http://127.0.0.1:9999', - *(['--docker-host-bridge'] if bridge else [])]) + run.main(args) receipt = (tmp_path / 'runs/route/fixture/runtime-receipt.json').read_text() assert 'synthetic-provider-secret' not in receipt + assert json.loads(receipt).get('provider_request_admission') == ('on' if provider_request_admission else None) assert json.loads(receipt).get('docker_host_bridge', False) is bridge diff --git a/benchmark/tests/test_shared_codex_runtime.py b/benchmark/tests/test_shared_codex_runtime.py index 91393f54d0..8e6cef4805 100644 --- a/benchmark/tests/test_shared_codex_runtime.py +++ b/benchmark/tests/test_shared_codex_runtime.py @@ -622,6 +622,48 @@ def test_baseline_and_treatment_use_same_harbor_entry(tmp_path): assert agent.replan_after_turns == (6 if agent.execution.uses_loopx else None) +@pytest.mark.parametrize("option,mode,provider_env", [ + ("on", "heartbeat", {}), + (1, "heartbeat", {}), + (None, "heartbeat", {}), + (True, "plain", {}), + (True, "native-goal", {}), + (True, "loopx-goal", {}), + (True, "turn", {}), + (True, "heartbeat", {"OPENAI_BASE_URL": ""}), + (True, "heartbeat", {"OPENAI_API_KEY": ""}), + (True, "heartbeat", {"CODEX_WIRE_API": "chat"}), +]) +def test_harbor_provider_request_admission_rejects_unsupported_activation( + tmp_path, option, mode, provider_env, +): + pytest.importorskip("harbor") + from benchmark.runtime.harbor import BenchmarkCodex + + with pytest.raises(ValueError, match="provider_request_admission"): + BenchmarkCodex( + logs_dir=tmp_path, model_name="fixture", execution_mode=mode, + provider_request_admission=option, + extra_env={"OPENAI_BASE_URL": "https://provider.invalid/v1", + "OPENAI_API_KEY": "synthetic-provider-key", + "CODEX_WIRE_API": "responses", **provider_env}, + **({"validation_command": ["true"]} if mode == "turn" else {}), + ) + + +def test_harbor_provider_request_admission_off_ignores_ambient_activation(tmp_path, monkeypatch): + pytest.importorskip("harbor") + from benchmark.runtime.harbor import BenchmarkCodex + + monkeypatch.setenv("LOOPX_PROVIDER_REQUEST_ADMISSION", "on") + kwargs = dict(logs_dir=tmp_path, model_name="fixture", + extra_env={"OPENAI_BASE_URL": "", "OPENAI_API_KEY": "", "CODEX_WIRE_API": "chat"}) + implicit = BenchmarkCodex(**kwargs) + explicit = BenchmarkCodex(**kwargs, provider_request_admission=False) + assert implicit._worker_env(cwd="/task") == explicit._worker_env(cwd="/task") + assert "LOOPX_PROVIDER_REQUEST_ADMISSION" not in implicit._worker_env(cwd="/task") + + def test_checkpoint_request_is_explicit_and_reaches_the_existing_worker_env(tmp_path, monkeypatch): pytest.importorskip("harbor") from benchmark.runtime.harbor import BenchmarkCodex @@ -825,7 +867,8 @@ def test_default_execution_has_no_independent_turn_deadline(tmp_path): (True, "fresh_replan", "explicit"), ]) @pytest.mark.parametrize("context_delivery", ["off", "on"]) -def test_harbor_install_receipt_keeps_capture_independent_from_display(tmp_path, monkeypatch, enabled, trigger, expected, context_delivery): +@pytest.mark.parametrize("provider_request_admission", [False, True]) +def test_harbor_install_receipt_keeps_capture_independent_from_display(tmp_path, monkeypatch, enabled, trigger, expected, context_delivery, provider_request_admission): pytest.importorskip("harbor") from benchmark.runtime.harbor import BenchmarkCodex, CodexOffline from benchmark.runtime.checkpoint_wake import CheckpointWake @@ -843,8 +886,12 @@ def test_harbor_install_receipt_keeps_capture_independent_from_display(tmp_path, request.write_text(json.dumps(declaration)) checkpoint_args = {"checkpoint_request_path": str(request), "checkpoint_request": request.read_bytes()} + provider_env = {"OPENAI_BASE_URL": "https://provider.invalid/v1", + "OPENAI_API_KEY": "synthetic-provider-key", "CODEX_WIRE_API": "responses"} agent = BenchmarkCodex(logs_dir=tmp_path, model_name="fixture", turn_envelope=enabled, context_delivery=context_delivery, + provider_request_admission=provider_request_admission, + extra_env=provider_env, **checkpoint_args) with CheckpointWake({"LOOPX_WAKE_LOG_DIR": str(wake_root), **({"LOOPX_CHECKPOINT_REQUEST": str(request)} if trigger else {})}, @@ -870,7 +917,7 @@ async def execute(*args, **kwargs): monkeypatch.setattr(agent, "_stage_source", stage) monkeypatch.setattr(agent, "exec_as_root", execute) monkeypatch.setattr(agent, "exec_as_agent", execute) - monkeypatch.setattr(agent, "_get_env", lambda key: None) + monkeypatch.setattr(agent, "_get_env", provider_env.get) async def task_exec(**kwargs): assert kwargs == {"command": 'printf "%s" "$PATH"', "timeout_sec": 30} return SimpleNamespace(stdout="/task/toolchain:/usr/bin:/bin", return_code=0) @@ -890,6 +937,12 @@ async def task_exec(**kwargs): assert receipt.get("turn_envelope_source") == expected assert receipt.get("context_delivery") == (context_delivery if context_delivery != "off" else None) assert agent._worker_env(cwd="/task").get("LOOPX_CONTEXT_DELIVERY") == ("on" if context_delivery != "off" else None) + assert receipt.get("provider_request_admission") == ("on" if provider_request_admission else None) + worker_env = agent._worker_env(cwd="/task") + assert worker_env.get("LOOPX_PROVIDER_REQUEST_ADMISSION") == ("on" if provider_request_admission else None) + assert worker_env["OPENAI_BASE_URL"] == provider_env["OPENAI_BASE_URL"] + assert worker_env["OPENAI_API_KEY"] == provider_env["OPENAI_API_KEY"] + assert provider_env["OPENAI_API_KEY"] not in json.dumps(receipt) if expected: assert receipt["turn_envelope"] is True diff --git a/loopx/capabilities/benchmark_toolkit/provider_gateway.py b/loopx/capabilities/benchmark_toolkit/provider_gateway.py index db39067622..2c8b5d14cf 100644 --- a/loopx/capabilities/benchmark_toolkit/provider_gateway.py +++ b/loopx/capabilities/benchmark_toolkit/provider_gateway.py @@ -10,9 +10,13 @@ import http.client import json +import math +import secrets +import socket import threading -from collections.abc import Iterator, Sequence -from contextlib import contextmanager +import time +from collections.abc import Callable, Iterator, Sequence +from contextlib import AbstractContextManager, contextmanager from dataclasses import dataclass from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from urllib.parse import SplitResult, urlsplit @@ -46,13 +50,21 @@ class RunnerOwnedProviderGateway: } -def _validated_upstream(value: str) -> SplitResult: +class ProviderGatewayRequestRejected(RuntimeError): + """A caller's current authority refused this request before dispatch.""" + + def __init__(self, code: str): + super().__init__(code) + self.code = code + + +def _validated_upstream(value: str, *, allow_http_upstream: bool = False) -> SplitResult: parsed = urlsplit(value.rstrip("/")) if parsed.scheme not in {"http", "https"} or not parsed.hostname: raise ValueError("upstream_base_url must be an absolute HTTP(S) URL") if parsed.username or parsed.password or parsed.query or parsed.fragment: raise ValueError("upstream_base_url must not contain credentials or query data") - if parsed.scheme == "http" and parsed.hostname not in { + if not allow_http_upstream and parsed.scheme == "http" and parsed.hostname not in { "127.0.0.1", "::1", "localhost", @@ -77,17 +89,59 @@ def _validated_allowed_paths(values: Sequence[str]) -> tuple[str, ...]: def _upstream_connection( - parsed: SplitResult, *, timeout_sec: float + parsed: SplitResult, *, timeout_sec: float, proxy_url: str | None = None ) -> http.client.HTTPConnection: host = parsed.hostname if host is None: # Defensive narrowing; _validated_upstream rejects this. raise ValueError("upstream_base_url must include a host") port = parsed.port + if proxy_url is not None: + proxy = urlsplit(proxy_url) + if (parsed.scheme != "https" or proxy.scheme != "http" or not proxy.hostname + or proxy.username or proxy.password or proxy.query or proxy.fragment + or proxy.path not in {"", "/"}): + raise ValueError("provider gateway requires a credential-free HTTP CONNECT proxy for HTTPS") + connection = http.client.HTTPSConnection(proxy.hostname, + proxy.port if proxy.port is not None else 80, timeout=timeout_sec) + connection.set_tunnel(host, port or 443) + return connection if parsed.scheme == "https": return http.client.HTTPSConnection(host, port, timeout=timeout_sec) return http.client.HTTPConnection(host, port, timeout=timeout_sec) +def _dispatch_bounded(connection: http.client.HTTPConnection, path: str, body: bytes, + headers: dict[str, str], deadline: float) -> None: + """Bound the entire admitted send, including real TCP backpressure. + + The connection (DNS, CONNECT and TLS included) is prepared outside the + caller's authority scope. No reconnect or automatic POST retry is allowed. + """ + connected_socket = connection.sock + if connected_socket is None: + raise OSError("provider connection closed before dispatch") + connection.auto_open = 0 + connection.putrequest("POST", path, + skip_accept_encoding=any(key.lower() == "accept-encoding" for key in headers)) + for key, value in headers.items(): + connection.putheader(key, value) + + def remaining() -> None: + budget = deadline - time.monotonic() + if not math.isfinite(budget) or budget <= 0: + raise TimeoutError("provider request dispatch deadline exceeded") + connected_socket.settimeout(budget) + + remaining() + connection.endheaders() + # Recompute the absolute budget between writes; repeated successful writes + # must not renew a socket timeout while a Goal writer is waiting. + view = memoryview(body) + for start in range(0, len(view), 64 * 1024): + remaining() + connection.send(view[start:start + 64 * 1024]) + + @contextmanager def serve_runner_owned_provider_gateway( *, @@ -96,6 +150,10 @@ def serve_runner_owned_provider_gateway( allowed_paths: Sequence[str] = _DEFAULT_ALLOWED_PATHS, request_timeout_sec: float = 600, max_request_bytes: int = 64 * 1024 * 1024, + request_lifetime: Callable[[], AbstractContextManager[float]] | None = None, + observe_request: Callable[[dict[str, object]], None] | None = None, + upstream_proxy_url: str | None = None, + allow_http_upstream: bool = False, ) -> Iterator[RunnerOwnedProviderGateway]: """Serve a loopback-only Responses API gateway for one benchmark run. @@ -104,17 +162,40 @@ def serve_runner_owned_provider_gateway( and the agent runtime. """ - upstream = _validated_upstream(upstream_base_url) + upstream = _validated_upstream(upstream_base_url, allow_http_upstream=allow_http_upstream) token = str(upstream_bearer_token) if not token.strip(): raise ValueError("upstream_bearer_token must be non-empty") if request_timeout_sec <= 0 or max_request_bytes <= 0: raise ValueError("gateway limits must be positive") admitted_paths = _validated_allowed_paths(allowed_paths) + if upstream_proxy_url is not None: + # Validate an explicit proxy before starting the native invocation. + _upstream_connection(upstream, timeout_sec=request_timeout_sec, + proxy_url=upstream_proxy_url).close() + # Each guarded server owns one invocation. A fresh unguessable path also + # prevents a stale URL from adopting a new invocation after port reuse. + prefix = "/" + secrets.token_hex(24) if request_lifetime is not None else "" + clients: set[socket.socket] = set() + clients_lock = threading.Lock() class GatewayHandler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.0" + def setup(self) -> None: + if request_lifetime is not None: + self.request.settimeout(request_timeout_sec) + with clients_lock: + clients.add(self.request) + super().setup() + + def finish(self) -> None: + try: + super().finish() + finally: + with clients_lock: + clients.discard(self.request) + def log_message(self, _format: str, *args: object) -> None: return @@ -131,7 +212,8 @@ def _json_error(self, status: int, reason: str) -> None: def do_POST(self) -> None: requested = urlsplit(self.path) - if requested.path not in admitted_paths: + path = requested.path.removeprefix(prefix) if prefix else requested.path + if (prefix and not requested.path.startswith(prefix + "/")) or path not in admitted_paths: self._json_error(404, "provider_gateway_path_not_admitted") return raw_length = self.headers.get("Content-Length") @@ -143,7 +225,10 @@ def do_POST(self) -> None: self._json_error(413, "provider_gateway_request_size_invalid") return body = self.rfile.read(content_length) - upstream_path = f"{upstream.path.rstrip('/')}{requested.path}" + if request_lifetime is not None and len(body) != content_length: + self._json_error(400, "provider_gateway_incomplete_request") + return + upstream_path = f"{upstream.path.rstrip('/')}{path}" if requested.query: upstream_path = f"{upstream_path}?{requested.query}" headers = { @@ -159,11 +244,34 @@ def do_POST(self) -> None: } headers["Authorization"] = f"Bearer {token}" headers["Content-Length"] = str(len(body)) - connection = _upstream_connection(upstream, timeout_sec=request_timeout_sec) + connection = _upstream_connection(upstream, timeout_sec=request_timeout_sec, + **({"proxy_url": upstream_proxy_url} if upstream_proxy_url is not None else {})) response_started = False + observed: dict[str, object] = {"path": path, "forwarded": False} + if observe_request is not None: + observed["request_id"] = secrets.token_hex(16) + + def observe(event: str) -> None: + if observe_request is not None: + observe_request({**observed, "event": event}) + try: - connection.request("POST", upstream_path, body=body, headers=headers) + observe("started") + if request_lifetime is None: + connection.request("POST", upstream_path, body=body, headers=headers) + else: + connection.connect() + with request_lifetime() as deadline: + observed["forwarded"] = None # A partial send has unknown acceptance. + _dispatch_bounded(connection, upstream_path, body, headers, deadline) + if connection.sock is not None: + connection.sock.settimeout(request_timeout_sec) + observed["forwarded"] = True # Bytes sent, not a provider acceptance receipt. + # Observation may write to disk. Keep it outside the admitted + # send scope and record dispatch before waiting for a response. + observe("dispatch") response = connection.getresponse() + observed["response_status"] = response.status self.send_response(response.status, response.reason) for key, value in response.getheaders(): if key.lower() not in { @@ -177,6 +285,9 @@ def do_POST(self) -> None: while chunk := response.read1(64 * 1024): self.wfile.write(chunk) self.wfile.flush() + except ProviderGatewayRequestRejected as error: + observed["refusal_code"] = error.code + self._json_error(409, error.code) except (OSError, http.client.HTTPException): if not response_started and not self.wfile.closed: self._json_error(502, "provider_gateway_upstream_unavailable") @@ -184,6 +295,7 @@ def do_POST(self) -> None: self.close_connection = True finally: connection.close() + observe("finished") def do_GET(self) -> None: self._json_error(405, "provider_gateway_method_not_admitted") @@ -200,12 +312,21 @@ def do_GET(self) -> None: host, port = server.server_address[:2] rendered_host = host.decode("ascii") if isinstance(host, bytes) else str(host) yield RunnerOwnedProviderGateway( - base_url=f"http://{rendered_host}:{port}", + base_url=f"http://{rendered_host}:{port}{prefix}", allowed_paths=admitted_paths, ) finally: server.shutdown() server.server_close() + if request_lifetime is not None: + # Interrupt old inbound waits on retirement. A handler already + # waiting for authority still sees its original retired owner. + with clients_lock: + for client in tuple(clients): + try: + client.shutdown(socket.SHUT_RDWR) + except OSError: + pass thread.join(timeout=5) diff --git a/tests/capabilities/test_provider_gateway_observation.py b/tests/capabilities/test_provider_gateway_observation.py new file mode 100644 index 0000000000..bd751f3b22 --- /dev/null +++ b/tests/capabilities/test_provider_gateway_observation.py @@ -0,0 +1,219 @@ +from __future__ import annotations + +import http.client +import json +import threading +from contextlib import contextmanager +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from types import SimpleNamespace +from urllib.parse import urlsplit + +from benchmark.runtime.provider_request_admission import ProviderRequestAdmission + + +class _Admission: + """Expose the lock ordering without introducing a second authority policy.""" + + def __init__(self): + self.lock = threading.Lock() + self.local = threading.local() + self.entries = 0 + self.pause_second = False + self.paused = threading.Event() + self.resume = threading.Event() + + def with_launch_decision(self, _check): + return self + + @contextmanager + def launch_lifetime(self, **_kwargs): + with self.lock: + self.entries += 1 + self.local.held = True + try: + if self.pause_second and self.entries == 2: + self.paused.set() + assert self.resume.wait(timeout=10) + yield + finally: + self.local.held = False + + +@contextmanager +def _waiting_upstream(): + received = threading.Event() + release = threading.Event() + observed = {} + + class Handler(BaseHTTPRequestHandler): + def log_message(self, _format, *args): + pass + + def do_POST(self): + observed["body"] = self.rfile.read(int(self.headers["Content-Length"])) + observed["accept_encoding"] = self.headers.get_all("Accept-Encoding") + received.set() + if release.wait(timeout=10): + body = b'{"ok":true}' + self.send_response(200) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + server.daemon_threads = True + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield f"http://127.0.0.1:{server.server_port}/v1", received, release, observed + finally: + release.set() + server.shutdown() + server.server_close() + thread.join(timeout=5) + + +def _guard(tmp_path, upstream, admission): + receipt = {} + session = SimpleNamespace(admission=admission, goal_ref={}, session_id="synthetic-session") + guard = ProviderRequestAdmission( + {"OPENAI_BASE_URL": upstream, "OPENAI_API_KEY": "synthetic-provider-key", + "LOOPX_TURN": "synthetic-turn"}, + session, receipt, tmp_path, + ) + guard.started(SimpleNamespace(poll=lambda: None)) + return guard, receipt + + +def _request(base_url): + result = {} + + def run(): + parsed = urlsplit(base_url) + connection = http.client.HTTPConnection(parsed.hostname, parsed.port, timeout=8) + try: + connection.request( + "POST", parsed.path + "/responses", body=b'{"model":"fixture"}', + headers={"Content-Type": "application/json", "Accept-Encoding": "gzip, deflate"}, + ) + response = connection.getresponse() + result.update(status=response.status, body=response.read()) + except Exception as error: + result["error"] = error + finally: + connection.close() + + thread = threading.Thread(target=run, daemon=True) + thread.start() + return thread, result + + +def test_provider_observes_dispatch_while_upstream_response_is_pending(tmp_path, monkeypatch): + with _waiting_upstream() as (upstream, received, release, observed): + admission = _Admission() + guard, receipt = _guard(tmp_path, upstream, admission) + dispatched = threading.Event() + finished = threading.Event() + observe = guard.observe + + def record(event): + assert not getattr(admission.local, "held", False) + observe(event) + if event.get("event") == "dispatch": + dispatched.set() + if event.get("event", "finished") == "finished": + finished.set() + + monkeypatch.setattr(guard, "observe", record) + with guard.serve(["codex", "exec", "-"]): + client, result = _request(guard.base_url) + try: + assert received.wait(timeout=5) + summary = receipt["provider_request_invocations"][0] + assert summary["requests"] == 1 + assert dispatched.wait(timeout=5) + assert summary["sent"] == 1 + assert summary["pending"] == 1 + assert summary["unknown"] == summary["refused"] == 0 + assert not finished.is_set() + release.set() + assert finished.wait(timeout=5) + finally: + release.set() + client.join(timeout=5) + assert not client.is_alive() + assert result == {"status": 200, "body": b'{"ok":true}'} + assert summary["requests"] == summary["sent"] == 1 + assert summary["pending"] == 0 + assert observed["accept_encoding"] == ["gzip, deflate"] + events = [json.loads(line) for line in (tmp_path / "provider-requests.jsonl").read_text().splitlines()] + assert [event["event"] for event in events] == ["started", "dispatch", "finished"] + assert len({event["request_id"] for event in events}) == 1 + assert "synthetic-provider-key" not in json.dumps(events) + + +def test_blocked_request_log_does_not_hold_invocation_or_admission_lock(tmp_path, monkeypatch): + log_blocked = threading.Event() + release_log = threading.Event() + open_file = Path.open + log_path = tmp_path / "provider-requests.jsonl" + + class SlowLog: + def __init__(self, stream): + self.stream = stream + + def write(self, line): + if json.loads(line).get("event", "finished") == "finished": + log_blocked.set() + assert release_log.wait(timeout=10) + return self.stream.write(line) + + @contextmanager + def delayed_open(path, *args, **kwargs): + with open_file(path, *args, **kwargs) as stream: + yield SlowLog(stream) if path == log_path and args == ("a",) else stream + + monkeypatch.setattr(Path, "open", delayed_open) + with _waiting_upstream() as (upstream, received, release_response, _observed): + admission = _Admission() + admission.pause_second = True + guard, _receipt = _guard(tmp_path, upstream, admission) + admitted = threading.Event() + writer_done = threading.Event() + + def next_request(): + with guard.lifetime(): + admitted.set() + + def writer(): + with admission.lock: + writer_done.set() + + next_thread = threading.Thread(target=next_request, daemon=True) + writer_thread = threading.Thread(target=writer, daemon=True) + with guard.serve(["codex", "exec", "-"]): + client, result = _request(guard.base_url) + try: + assert received.wait(timeout=5) + # The next admission has passed its initial state check and now + # owns the canonical lock. Finish the first response into a slow log. + next_thread.start() + assert admission.paused.wait(timeout=5) + release_response.set() + assert log_blocked.wait(timeout=5) + admission.resume.set() + writer_thread.start() + assert writer_done.wait(timeout=5) + assert admitted.is_set() + assert not release_log.is_set() + finally: + admission.resume.set() + release_log.set() + release_response.set() + client.join(timeout=5) + if next_thread.ident is not None: + next_thread.join(timeout=5) + if writer_thread.ident is not None: + writer_thread.join(timeout=5) + assert result == {"status": 200, "body": b'{"ok":true}'} diff --git a/tests/capabilities/test_provider_gateway_proxy.py b/tests/capabilities/test_provider_gateway_proxy.py new file mode 100644 index 0000000000..2372f7d808 --- /dev/null +++ b/tests/capabilities/test_provider_gateway_proxy.py @@ -0,0 +1,139 @@ +from __future__ import annotations + +from contextlib import contextmanager +import http.client +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import json +import threading +import time +from urllib.parse import urlsplit + +import pytest + +from loopx.capabilities.benchmark_toolkit.provider_gateway import ( + ProviderGatewayRequestRejected, + _upstream_connection, + serve_runner_owned_provider_gateway, +) + + +@pytest.mark.parametrize("proxy_url,proxy_port", [ + ("http://proxy.invalid", 80), + ("http://proxy.invalid:3128", 3128), +]) +@pytest.mark.parametrize("upstream_url,target_port", [ + ("https://provider.invalid/v1", 443), + ("https://provider.invalid:8443/v1", 8443), +]) +def test_http_connect_proxy_port_does_not_change_https_tunnel_target( + proxy_url, proxy_port, upstream_url, target_port, +): + connection = _upstream_connection( + urlsplit(upstream_url), timeout_sec=2, proxy_url=proxy_url, + ) + try: + assert isinstance(connection, http.client.HTTPSConnection) + assert connection.host == "proxy.invalid" + assert connection.port == proxy_port + assert connection._tunnel_host == "provider.invalid" + assert connection._tunnel_port == target_port + assert connection.sock is None, "route validation must not connect" + finally: + connection.close() + + +@pytest.fixture +def blocked_connect_proxy(): + connected, release = threading.Event(), threading.Event() + targets = [] + + class Proxy(BaseHTTPRequestHandler): + def log_message(self, *args): + pass + + def do_CONNECT(self): # noqa: N802 - BaseHTTPRequestHandler contract + targets.append(self.path) + connected.set() + release.wait(timeout=10) + # No public endpoint or TLS service is needed: exercise a real + # blocked CONNECT, then return an ordinary proxy failure. + self.send_response(502) + self.send_header("Content-Length", "0") + self.end_headers() + + server = ThreadingHTTPServer(("127.0.0.1", 0), Proxy) + server.daemon_threads = True + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield f"http://127.0.0.1:{server.server_port}", connected, release, targets + finally: + release.set() + server.shutdown() + server.server_close() + thread.join(timeout=5) + assert not thread.is_alive() + + +def test_slow_connect_leaves_dispatch_lifetime_free_for_stop_writer(blocked_connect_proxy): + proxy_url, connected, release, targets = blocked_connect_proxy + authority_lock = threading.Lock() + admission_entered, stop_committed = threading.Event(), threading.Event() + response = {} + + @contextmanager + def lifetime(): + # A transport-level oracle: the same lock protects dispatch admission + # and its Stop writer. This does not substitute for canonical Goal tests. + with authority_lock: + admission_entered.set() + if stop_committed.is_set(): + raise ProviderGatewayRequestRejected("fixture_stop_committed") + yield time.monotonic() + 2 + + def commit_stop(): + with authority_lock: + stop_committed.set() + + with serve_runner_owned_provider_gateway( + upstream_base_url="https://provider.invalid/v1", + upstream_bearer_token="synthetic-upstream-credential", + upstream_proxy_url=proxy_url, + request_timeout_sec=5, + request_lifetime=lifetime, + ) as gateway: + target = urlsplit(gateway.base_url) + + def request(): + client = http.client.HTTPConnection(target.hostname, target.port, timeout=6) + try: + client.request("POST", target.path + "/responses", body=b"{}", + headers={"Content-Type": "application/json"}) + result = client.getresponse() + response.update(status=result.status, body=json.loads(result.read())) + except Exception as error: + response["error"] = error + finally: + client.close() + + requester = threading.Thread(target=request, daemon=True) + writer = threading.Thread(target=commit_stop, daemon=True) + requester.start() + try: + assert connected.wait(timeout=5), "the actual CONNECT did not reach the local proxy" + writer.start() + assert stop_committed.wait(timeout=2), "CONNECT held the dispatch authority lock" + assert not release.is_set(), "Stop must commit before the proxy is released" + assert not admission_entered.is_set() + assert targets == ["provider.invalid:443"] + finally: + release.set() + requester.join(timeout=6) + if writer.ident is not None: + writer.join(timeout=2) + assert not requester.is_alive() + assert not writer.is_alive() + assert "error" not in response, response + assert response["status"] == 502 + assert response["body"]["error"]["code"] == "provider_gateway_upstream_unavailable" + assert not admission_entered.is_set(), "a failed CONNECT must never enter admission" diff --git a/tests/control_plane/test_benchmark_checkpoint_admission.py b/tests/control_plane/test_benchmark_checkpoint_admission.py index 1cf507f441..fc78bf4865 100644 --- a/tests/control_plane/test_benchmark_checkpoint_admission.py +++ b/tests/control_plane/test_benchmark_checkpoint_admission.py @@ -39,6 +39,12 @@ with Path("checkpoint-model.jsonl").open("a") as stream: stream.write(json.dumps({"turn":env["LOOPX_TURN"]})+"\n") print(json.dumps({"type":"thread.started", "thread_id":"benchmark-original-session"}), flush=True) +if env.get("LOOPX_PROVIDER_REQUEST_ADMISSION") == "on": + import urllib.request + override = next(arg for arg in sys.argv if arg.startswith("model_providers.harbor.base_url=")) + endpoint = json.loads(override.split("=", 1)[1]) + with urllib.request.urlopen(urllib.request.Request(endpoint + "/responses", data=b"{}"), timeout=30) as response: + assert response.status == 200 base = [env["LOOPX_CLI"], "--format", "json", "--registry", env["LOOPX_REGISTRY"], "--runtime-root", env["LOOPX_RUNTIME_ROOT"]] def cli(*args): @@ -147,13 +153,34 @@ def test_checkpoint_unbound_guard_selects_existing_work_in_original_turn(tmp_pat _run_checkpoint(tmp_path, monkeypatch, provider, refusal=None, todo_bound=False, unbound=True) -def _run_checkpoint(tmp_path, monkeypatch, provider, *, refusal, todo_bound, unbound=False, workspace_change=False, work_bound=False): +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_checkpoint_request_admission_creates_a_new_invocation_for_original_turn(tmp_path, monkeypatch, provider): + from tests.capabilities.test_provider_gateway import _recording_upstream + + with _recording_upstream() as (upstream, observed): + receipt, native, home = _run_checkpoint(tmp_path, monkeypatch, provider, + refusal=None, todo_bound=False, provider_url=upstream) + assert observed["path"] == "/v1/responses" + invocations = receipt["provider_request_invocations"] + assert [row["turn_id"] for row in invocations] == [OUTER_TURN, HELD_TURN] + assert len({row["invocation_id"] for row in invocations}) == 2 + assert all(row["requests"] == row["sent"] == 1 and row["refused"] == 0 for row in invocations) + endpoints = [next(arg for arg in row["argv"] if arg.startswith("model_providers.harbor.base_url=")) + for row in native] + assert endpoints[0] != endpoints[1] + assert upstream in home.read_text() + assert all(endpoint.split("=", 1)[1] not in home.read_text() for endpoint in endpoints) + + +def _run_checkpoint(tmp_path, monkeypatch, provider, *, refusal, todo_bound, unbound=False, workspace_change=False, work_bound=False, provider_url=None): unbound = unbound or refusal == "original_event_changed" env, project, runtime, source, state = _environment( tmp_path, monkeypatch, provider, replan="todo" if todo_bound or unbound else not work_bound) env["CODEX_BIN"] = str(write_python_host_fixture(project / "checkpoint-host", f"#!{sys.executable}\n" + _HOST)) env = process_environment(Path(env["LOOPX_CODEX_HOME"]), base=env) env.update(LOOPX_TURN=OUTER_TURN, FIXTURE_OUTER_TURN=OUTER_TURN, FIXTURE_HELD_TURN=HELD_TURN) + if provider_url is not None: + env.update(OPENAI_BASE_URL=provider_url, LOOPX_PROVIDER_REQUEST_ADMISSION="on") if unbound: env["FIXTURE_SELECTION_TODO"] = fixture.SELECTED_REPLAN_TODO_ID if todo_bound: @@ -291,3 +318,4 @@ def captured_then_changed(held, deadline): assert events[:len(observations["events"])] == observations["events"] if not unbound: assert events == observations["events"] + return receipt, native, Path(env["LOOPX_CODEX_HOME"]) / "config.toml" diff --git a/tests/control_plane/test_benchmark_request_worker.py b/tests/control_plane/test_benchmark_request_worker.py new file mode 100644 index 0000000000..eed00ebc9f --- /dev/null +++ b/tests/control_plane/test_benchmark_request_worker.py @@ -0,0 +1,98 @@ +"""Real worker/canonical admission with a scripted host and local HTTP. + +These tests do not qualify a Codex binary or invoke a model. +""" +from __future__ import annotations + +import json +from pathlib import Path +import sys +import time +from types import SimpleNamespace + +import pytest + +from benchmark.runtime import worker +from loopx.control_plane.goals.first_party_host_admission import FirstPartyHostRuntimeRejected +from tests.capabilities.test_provider_gateway import _recording_upstream +from tests.control_plane.host_process_fixture import write_python_host_fixture +from tests.control_plane.test_benchmark_current_admission import _environment + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("enabled", [False, True]) +def test_worker_request_route_is_invocation_only_and_default_off(tmp_path, monkeypatch, provider, enabled): + env, project, _runtime, _source, _state = _environment(tmp_path, monkeypatch, provider) + host = r''' +import json, os, sys, urllib.request +from pathlib import Path +sys.stdin.read() +print(json.dumps({"type":"thread.started", "thread_id":"benchmark-original-session"}), flush=True) +overrides = [arg for arg in sys.argv if arg.startswith("model_providers.harbor.base_url=")] +endpoint = json.loads(overrides[0].split("=",1)[1]) if overrides else os.environ["OPENAI_BASE_URL"] +with urllib.request.urlopen(urllib.request.Request(endpoint + "/responses",data=b'{"input":"fixture"}'), timeout=30) as response: + assert response.status == 200 +Path("request-route.json").write_text(json.dumps({"endpoint":endpoint,"argv":sys.argv[1:]})) +''' + env["CODEX_BIN"] = str(write_python_host_fixture(project / "provider-host", f"#!{sys.executable}\n" + host)) + with _recording_upstream() as (upstream, observed): + env["OPENAI_BASE_URL"] = upstream + if enabled: + env["LOOPX_PROVIDER_REQUEST_ADMISSION"] = "on" + receipt = worker.run_once(env) + assert receipt["ok"] and receipt["host_invoked"] + assert observed["path"] == "/v1/responses" and observed["body"] == b'{"input":"fixture"}' + route = json.loads((project / "request-route.json").read_text()) + home = (Path(env["LOOPX_CODEX_HOME"]) / "config.toml").read_text() + assert upstream in home + if enabled: + assert route["endpoint"] != upstream and route["endpoint"] not in home + assert receipt["provider_request_admission"] == "on" + [invocation] = receipt["provider_request_invocations"] + assert invocation["turn_id"] == receipt["turn_id"] + assert invocation["requests"] == invocation["sent"] == 1 + assert invocation["refused"] == invocation["unknown"] == 0 + else: + assert route["endpoint"] == upstream + assert "provider_request_admission" not in receipt + assert "provider_request_invocations" not in receipt + + +def test_worker_does_not_certify_native_success_after_request_refusal(tmp_path, monkeypatch): + from benchmark.runtime import provider_request_admission + + env, project, _runtime, _source, _state = _environment(tmp_path, monkeypatch, "sqlite") + # The original worker/launch check is real. Only the transport deadline is + # exhausted after native attachment, so Popen has already succeeded. + original = provider_request_admission.ProviderRequestAdmission.started + clock = [time.monotonic()] + monkeypatch.setattr(provider_request_admission, "time", SimpleNamespace( + time=time.time, monotonic=lambda: clock[0])) + + def exhausted(self, process): + original(self, process) + clock[0] += 3600 + + monkeypatch.setattr(provider_request_admission.ProviderRequestAdmission, "started", exhausted) + host = r''' +import json, signal, sys, urllib.error, urllib.request +signal.signal(signal.SIGINT, signal.SIG_IGN) +sys.stdin.read() +print(json.dumps({"type":"thread.started","thread_id":"benchmark-original-session"}),flush=True) +endpoint = json.loads(next(arg for arg in sys.argv if arg.startswith("model_providers.harbor.base_url=")).split("=",1)[1]) +try: + urllib.request.urlopen(urllib.request.Request(endpoint+"/responses",data=b"{}"),timeout=30) +except urllib.error.HTTPError as error: + assert error.code == 409 +sys.exit(0) +''' + env["CODEX_BIN"] = str(write_python_host_fixture(project / "provider-host", f"#!{sys.executable}\n" + host)) + with _recording_upstream() as (upstream, observed): + env.update(OPENAI_BASE_URL=upstream, LOOPX_PROVIDER_REQUEST_ADMISSION="on") + with pytest.raises(FirstPartyHostRuntimeRejected, match="provider_request_budget_exhausted"): + worker.run_once(env) + [receipt_path] = Path(env["LOOPX_WAKE_LOG_DIR"]).glob("*/receipt.json") + receipt = json.loads(receipt_path.read_text()) + assert not receipt["ok"] and receipt["return_code"] == 0 + assert receipt["host_failure"]["kind"] == "unknown" and not receipt["host_failure"]["retryable"] + assert not observed