diff --git a/benchmark/README.md b/benchmark/README.md index e110d289f0..f0d286fd8c 100644 --- a/benchmark/README.md +++ b/benchmark/README.md @@ -146,9 +146,18 @@ still owns endpoint access, actual model/tool qualification and private evidence With `OPENAI_BASE_URL` unset, login-file behavior is unchanged, including the required `CODEX_AUTH_JSON_PATH`; an ambient API key alone does not select the -custom route. Change routes only between fresh attempts. Checkpoint continuation -must retain the captured endpoint/configuration and supply fresh credentials -externally. This routing repair enables matched experiments under the +custom route. Change providers only between separately admitted attempts. +Checkpoint continuation retains the sealed source configuration and supplies +fresh credentials externally. For a fresh SForge receiver, the Host binds its +current credential-gateway URL before native session admission; an ephemeral +gateway port is transport, not a reason to reuse a dead source route. The runner +renders the destination with the ordinary worker's settings owner and verifies +that only this route and any explicitly selected model/effort differ. Both full +configuration digests and session profiles remain bound to the new receiver. +Ordinary within-trial configuration changes still reject. Endpoint access and +the same upstream provider remain the operator's separately verified contract; +a URL or configuration digest does not prove provider identity or quality. +This routing repair enables matched experiments under the [long-horizon research contract](../docs/architecture/rfcs/long-horizon-harness-benchmark-research-program-v0.md); it establishes neither sustained operation nor a solver-quality improvement. @@ -1099,7 +1108,14 @@ also match `--model`/`--effort`. The native session participant records both configuration digests, verifies the original session profile and full retained history, then binds the new profile to a distinct destination session. Retry uses that same source/destination intent; it cannot switch an already running -receiver. Without the option, exact-configuration admission is unchanged. +receiver. Direct Host admission without the option still requires exact configuration. +The SForge Host may additionally rebind its existing custom provider's +credential-gateway URL from the current launch environment. This applies only +to a fresh checkpoint receiver: the provider name, wire API, credential source, +tools and permission settings must stay fixed. It adds no operator option or +permission to edit an active receiver. The original source configuration stays +sealed; admission records the full source/destination configuration and profile +digests through the existing transition provenance. The host controller exposes the same explicit transition as `--destination-model-config`; this remains a CLI/Host journey, with frontend restore support still open. Changing the model changes the experiment treatment: @@ -1115,6 +1131,12 @@ It grants no extra native submissions, cooldown relief, permission or time. `--destination-model-config`；frontend restore 旅程仍未交付。模型变化是新的实验因素， 共享 checkpoint 本身不能证明机制收益。 +SForge 宿主可在全新 checkpoint 接收端准入前，将既有自定义 provider 的凭据网关 +URL 绑定到当前启动环境；临时端口变化不应迫使接收端继续使用已失效的源网关。 +Provider 名称、wire API、凭据来源、工具和权限保持一致，完整源配置仍封存， +既有 transition 记录保留源与目标配置和 session profile 摘要。此处没有新增操作选项， +也不允许修改活动接收端；同一上游 provider 的资格仍需独立核验。 + Warm verification hashes the complete checkpoint with a read window of up to 60 seconds, capped by the original execution deadline. It rejects a read that finishes after that deadline. The launcher reuses this verified metadata before diff --git a/benchmark/runtime/codex.py b/benchmark/runtime/codex.py index 30acca4b5d..35980340a1 100644 --- a/benchmark/runtime/codex.py +++ b/benchmark/runtime/codex.py @@ -115,46 +115,17 @@ def validate_model_catalog(model_catalog: Path, model: str) -> dict: ) return entry -def prepare_codex_home( - home: Path, - *, - execution: Execution, - workspace: Path, - model: str, - effort: str, - base_url: str, - api_key: str, - wire_api: str, - skills: Path | None, - web_search: str = "disabled", - model_catalog: Path | None = None, - model_catalog_destination: Path | None = None, -) -> None: - """Materialize fixed inputs once; never delete sessions between iterations. +def codex_settings( + *, execution: Execution, workspace: Path, model: str, effort: str, + base_url: str, wire_api: str, web_search: str = "disabled", + model_catalog: Path | None = None, model_catalog_destination: Path | None = None, +) -> str: + """Render fixed settings for the worker and its fresh checkpoint receiver. - A different effective configuration requires a different trial. Secrets are - kept out of the settings identity and may rotate without resetting history. - - ``model_catalog`` is required for slugs Codex does not ship built-in - (e.g. gateway aliases such as ``gpt-5.6-sol``). Without a catalog entry - Codex treats the model as unknown and offers a reduced tool surface, and - the ``web_search`` setting has no effect. A catalog entry is not - capability-only: it can select a tool mode and carry ``model_messages`` - that become prompt content. Entries whose tool mode would strip the - function-tool surface (for example ``code_mode_only``, which makes the - real client send ``tools: []``) are rejected before a trial; qualify a - pinned binary/catalog pair at the first-request boundary with - ``benchmark.runtime.toolface`` before relying on the rendered tools. - Host-side installers may set ``model_catalog_destination`` to the absolute - guest path where those validated catalog bytes are staged. It affects only - rendering; the host source must still pass the same validation. + Rendering does not create a home, change a session or handle credentials. """ if not model or not effort: raise ValueError("model and effort are required") - if base_url and not api_key: - raise ValueError("custom provider requires OPENAI_API_KEY") - if not base_url and not (home / "auth.json").is_file(): - raise ValueError("provide a gateway and API key, or stage CODEX_AUTH_JSON_PATH") if wire_api not in {"responses", "chat"}: raise ValueError("unsupported provider wire_api") if web_search not in {"disabled", "cached", "live"}: @@ -165,8 +136,7 @@ def prepare_codex_home( model_catalog is None or not model_catalog_destination.is_absolute() ): raise ValueError("catalog destination requires a validated catalog and an absolute guest path") - home.mkdir(parents=True, exist_ok=True) - settings = "\n".join( + return "\n".join( [ f"model = {json.dumps(model)}", f"model_reasoning_effort = {json.dumps(effort)}", @@ -213,6 +183,54 @@ def prepare_codex_home( ) + [""] ) + + +def prepare_codex_home( + home: Path, + *, + execution: Execution, + workspace: Path, + model: str, + effort: str, + base_url: str, + api_key: str, + wire_api: str, + skills: Path | None, + web_search: str = "disabled", + model_catalog: Path | None = None, + model_catalog_destination: Path | None = None, +) -> None: + """Materialize fixed inputs once; never delete sessions between iterations. + + A different effective configuration requires a different trial. Secrets are + kept out of the settings identity and may rotate without resetting history. + + ``model_catalog`` is required for slugs Codex does not ship built-in + (e.g. gateway aliases such as ``gpt-5.6-sol``). Without a catalog entry + Codex treats the model as unknown and offers a reduced tool surface, and + the ``web_search`` setting has no effect. A catalog entry is not + capability-only: it can select a tool mode and carry ``model_messages`` + that become prompt content. Entries whose tool mode would strip the + function-tool surface (for example ``code_mode_only``, which makes the + real client send ``tools: []``) are rejected before a trial; qualify a + pinned binary/catalog pair at the first-request boundary with + ``benchmark.runtime.toolface`` before relying on the rendered tools. + Host-side installers may set ``model_catalog_destination`` to the absolute + guest path where those validated catalog bytes are staged. It affects only + rendering; the host source must still pass the same validation. + """ + if not model or not effort: + raise ValueError("model and effort are required") + if base_url and not api_key: + raise ValueError("custom provider requires OPENAI_API_KEY") + if not base_url and not (home / "auth.json").is_file(): + raise ValueError("provide a gateway and API key, or stage CODEX_AUTH_JSON_PATH") + settings = codex_settings( + execution=execution, workspace=workspace, model=model, effort=effort, + base_url=base_url, wire_api=wire_api, web_search=web_search, + model_catalog=model_catalog, model_catalog_destination=model_catalog_destination, + ) + home.mkdir(parents=True, exist_ok=True) config = home / "config.toml" if config.exists() and config.read_text(encoding="utf-8") != settings: raise ValueError("Codex settings changed within a trial; use a new trial") diff --git a/benchmark/runtime/sforge.py b/benchmark/runtime/sforge.py index a9b983f4f9..1e9abdbcab 100644 --- a/benchmark/runtime/sforge.py +++ b/benchmark/runtime/sforge.py @@ -20,7 +20,7 @@ from benchmark.edgebench.feedback import FEEDBACK_MODES from benchmark.edgebench.feedback_hook import FEEDBACK_PAYLOAD -from .codex import DEFAULT_REPLAN_AFTER_TURNS, Execution, prepare_codex_home +from .codex import DEFAULT_REPLAN_AFTER_TURNS, Execution, codex_settings, prepare_codex_home from .codex_offline import CodexOffline, MODEL_CATALOG_REMOTE, stage_portable_python from .harbor import ( BenchmarkCodex, _CONTROL, _CODEX_HOME, _PYTHON, _SCHEDULER_STATE, _SRC, @@ -317,7 +317,24 @@ def _install_worker(self, backend, handle, log_dir, logger): ) asyncio.run(self.runtime.install(self.environment)) if self.warm_start is not None: - research_paths = asyncio.run(self.warm_start.stage(self.environment, self.cwd)) + route_settings = {} + if runtime_base_url: + worker_env = self.runtime._worker_env(cwd=self.cwd) + route_settings = { + "provider_base_url": runtime_base_url, + "effective_config": codex_settings( + execution=self.runtime.execution, workspace=Path(self.cwd), + model=worker_env["MODEL_NAME"], effort=worker_env["REASONING_EFFORT"], + base_url=runtime_base_url, + wire_api=worker_env["CODEX_WIRE_API"], + web_search=worker_env["LOOPX_WEB_SEARCH"], + model_catalog=self.runtime._model_catalog_source, + model_catalog_destination=(Path(MODEL_CATALOG_REMOTE) + if self.runtime._model_catalog_staged else None), + ).encode("utf-8"), + } + research_paths = asyncio.run(self.warm_start.stage( + self.environment, self.cwd, **route_settings)) self._warm_command("workspace", user="root") result = backend.exec_run(handle, ["chown", "-R", "-P", "agent:agent", self.cwd, *research_paths], user="root") diff --git a/benchmark/runtime/sforge_warm_start.py b/benchmark/runtime/sforge_warm_start.py index ecccaa0f30..f70155500f 100644 --- a/benchmark/runtime/sforge_warm_start.py +++ b/benchmark/runtime/sforge_warm_start.py @@ -16,6 +16,7 @@ import shutil import tempfile import time +import tomllib from loopx.control_plane.effect_runtime import effect_runtime_result from benchmark.runtime.layout import _CODEX_HOME, _CONTROL, _SRC @@ -96,14 +97,41 @@ def _verify(self, workspace: str) -> dict: raise ValueError("Original warm start deadline expired during checkpoint verification") return metadata - async def stage(self, environment, workspace: str) -> list[str]: + async def stage(self, environment, workspace: str, *, + effective_config: bytes | None = None, + provider_base_url: str | None = None) -> list[str]: from loopx.hosts.checkpoint_retention import checkpoint_bank_use with checkpoint_bank_use(self.root): - return await self._stage(environment, workspace) + return await self._stage(environment, workspace, + effective_config=effective_config, provider_base_url=provider_base_url) - async def _stage(self, environment, workspace: str) -> list[str]: + async def _stage(self, environment, workspace: str, *, + effective_config: bytes | None = None, + provider_base_url: str | None = None) -> list[str]: metadata = self._verify(workspace) research = _research_components(self.root, metadata, Path(workspace), self.research_component_ids) + requested = self.destination_config or self.config + transport_config = None + if effective_config is not None: + from loopx.hosts.codex_checkpoint import checkpoint_destination_config + + # The runner renders its actual launch through the ordinary worker's + # settings owner. Never change the operator's model/effort selection + # merely because a destination was rendered successfully. + before = tomllib.loads(requested.read_text(encoding="utf-8")) + after = tomllib.loads(effective_config.decode("utf-8")) + if any(before.get(key) != after.get(key) for key in ("model", "model_reasoning_effort")): + raise ValueError("Checkpoint launch differs from its declared model and reasoning effort") + with tempfile.TemporaryDirectory(prefix="checkpoint-route-") as folder: + target = Path(folder).resolve() / "config.toml" + target.write_bytes(effective_config) + checkpoint_destination_config(self.config, + metadata["native_session"]["config_sha256"], target, + provider_base_url=provider_base_url) + if effective_config != requested.read_bytes(): + transport_config = effective_config + elif provider_base_url is not None and effective_config is None: + raise ValueError("Checkpoint Host route requires its rendered destination settings") # Install tools first without creating the native home. Stage inert # participants outside the workspace so restoring its bytes cannot # replace the active controller, registry or installed executable. @@ -115,7 +143,16 @@ async def _stage(self, environment, workspace: str) -> list[str]: source_todo_id=self.source_todo_id, deadline_epoch=self.deadline_epoch, normal_cadence=self.normal_cadence) protected_files = [config, f"{_CONTROL}/warm-start.json"] - if self.destination_config is not None: + if transport_config is not None: + target = f"{_CONTROL}/restored-model-config.toml" + with tempfile.TemporaryDirectory(prefix="checkpoint-route-") as folder: + path = Path(folder) / "config.toml" + path.write_bytes(transport_config) + await environment.upload_file(path, target) + payload["destination_config"] = target + payload["provider_base_url"] = provider_base_url + protected_files.append(target) + elif self.destination_config is not None: target = f"{_CONTROL}/restored-model-config.toml" await environment.upload_file(self.destination_config, target) payload["destination_config"] = target @@ -302,7 +339,9 @@ def admit_warm_session(env: dict[str, str], declaration: dict) -> dict: result = admit_native_checkpoint_session(env, Path(declaration["root"]), declaration["sha256"], Path(declaration["config"]), sandbox=env["LOOPX_CODEX_SANDBOX"], timeout=timeout, **({"destination_config": Path(declaration["destination_config"])} - if declaration.get("destination_config") is not None else {})) + if declaration.get("destination_config") is not None else {}), + **({"provider_base_url": declaration["provider_base_url"]} + if declaration.get("provider_base_url") is not None else {})) _phase_read_timeout(declaration, 90) return result diff --git a/benchmark/tests/test_checkpoint_model_transition.py b/benchmark/tests/test_checkpoint_model_transition.py index e85ef5585f..3781c593ba 100644 --- a/benchmark/tests/test_checkpoint_model_transition.py +++ b/benchmark/tests/test_checkpoint_model_transition.py @@ -60,6 +60,61 @@ def test_model_transition_requires_an_explicit_model(settings, replacement): checkpoint_destination_config(source, digest, target) +@pytest.fixture +def gateway_settings(tmp_path): + source, target = tmp_path / "source.toml", tmp_path / "target.toml" + source.write_text('model="original"\nmodel_reasoning_effort="high"\n' + 'model_provider="gateway"\nsandbox_mode="read-only"\n' + '[model_providers.gateway]\nbase_url="http://gateway:4001/v1"\n' + 'wire_api="responses"\nenv_key="OPENAI_API_KEY"\n' + '[model_providers.other]\nbase_url="https://other.invalid/v1"\n') + target.write_text(source.read_text().replace(':4001/', ':4002/')) + return source, target, hashlib.sha256(source.read_bytes()).hexdigest() + + +def test_gateway_rebind_is_explicit_and_retains_full_configuration_provenance(gateway_settings): + source, target, digest = gateway_settings + before = source.read_bytes() + with pytest.raises(ValueError, match="only model and reasoning effort"): + checkpoint_destination_config(source, digest, target) + data, transition = checkpoint_destination_config(source, digest, target, + provider_base_url="http://gateway:4002/v1") + assert data == target.read_bytes() and source.read_bytes() == before + assert transition["source_config_sha256"] == digest + assert transition["destination_config_sha256"] == hashlib.sha256(data).hexdigest() + assert transition["source_model"] == transition["destination_model"] == "original" + + +@pytest.mark.parametrize("old,new", [ + ('wire_api="responses"', 'wire_api="chat"'), + ('env_key="OPENAI_API_KEY"', 'env_key="OTHER_KEY"'), + ('sandbox_mode="read-only"', 'sandbox_mode="danger-full-access"'), + ('https://other.invalid/v1', 'https://redirect.invalid/v1'), + ('model_provider="gateway"', 'model_provider="other"'), +]) +def test_gateway_rebind_cannot_change_other_inputs(gateway_settings, old, new): + source, target, digest = gateway_settings + target.write_text(target.read_text().replace(old, new)) + with pytest.raises(ValueError): + checkpoint_destination_config(source, digest, target, + provider_base_url="http://gateway:4002/v1") + + +def test_gateway_rebind_requires_matching_destination_and_current_launch(gateway_settings, monkeypatch): + from loopx.hosts import codex_checkpoint as owner + + source, target, digest = gateway_settings + with pytest.raises(ValueError, match="current Host route"): + checkpoint_destination_config(source, digest, target, provider_base_url="http://foreign:4002/v1") + with pytest.raises(ValueError, match="explicit destination settings"): + checkpoint_destination_config(source, digest, provider_base_url="http://gateway:4002/v1") + monkeypatch.setattr(owner, "fork_native_checkpoint", lambda *a, **kw: pytest.fail("must not fork")) + with pytest.raises(ValueError, match="current launch"): + owner.admit_native_checkpoint_session({"OPENAI_BASE_URL": "http://foreign:4002/v1"}, + source.parent, digest, source, sandbox="read-only", timeout=30, + destination_config=target, provider_base_url="http://gateway:4002/v1") + + def test_transition_provenance_survives_exact_session_observations_and_qualified_refresh(tmp_path): from loopx.control_plane.turn_driver.codex_sessions import ( _replace_checkpoint_codex_session, _store_codex_cli_session, load_codex_cli_session, diff --git a/benchmark/tests/test_native_checkpoint.py b/benchmark/tests/test_native_checkpoint.py index 09dbf8db14..60e0d9b1eb 100644 --- a/benchmark/tests/test_native_checkpoint.py +++ b/benchmark/tests/test_native_checkpoint.py @@ -306,7 +306,7 @@ def test_real_native_workspace_readback_rejects_foreign_code_basis(tmp_path): reason="Requires an explicit real native Host binary; only a local static provider is used") @pytest.mark.parametrize("case", ["resume", "scope_change", "profile_change", "joint", "binding_reply_loss", "fork_reply_loss", "fork_missing", "fork_ambiguous", "warm_preparation", - "model_transition"]) + "model_transition", "provider_transport", "joint_provider_transport"]) def test_real_fork_admission_and_native_resume_use_retained_state(tmp_path, capsys, monkeypatch, case, request): """Real Host/CLI/Goal/session/artifact owners; deterministic synthetic work. @@ -319,6 +319,8 @@ def test_real_fork_admission_and_native_resume_use_retained_state(tmp_path, caps import shlex import subprocess import sys + import tomllib + from contextlib import ExitStack from pathlib import Path from benchmark.runtime.codex import Execution from loopx.hosts.codex_checkpoint import admit_native_checkpoint_session @@ -330,6 +332,8 @@ def test_real_fork_admission_and_native_resume_use_retained_state(tmp_path, caps from loopx.control_plane.quota.settlement import read_heartbeat_settlement from loopx.control_plane.projects.registry_codec import ProjectRegistryProtocolError + joint = case in {"joint", "joint_provider_transport"} + provider_transport = case in {"provider_transport", "joint_provider_transport"} workspace = tmp_path / "workspace" workspace.mkdir() monkeypatch.chdir(workspace) # Real settlement verifies the producing workspace. @@ -387,7 +391,7 @@ def invoke(arguments): requests = [] retained_context_path = None joint_command = "" - if case == "joint": + if joint: import copy import shlex import requests as http @@ -424,7 +428,7 @@ def response(request, index, message): return {"type": "function_call", "id": "fc_retained", "call_id": "call_retained", "name": "exec_command", "arguments": json.dumps({ "cmd": read_current + "cat candidate.txt research.txt " + retained_context_path + joint_command, - "max_output_tokens": 6000})} + "yield_time_ms": 10000, "max_output_tokens": 6000})} if index == 3: message["content"][0]["text"] = "Retained candidate 7 verified; candidate 3 remains rejected." return message @@ -432,10 +436,10 @@ def response(request, index, message): binary = os.environ["LOOPX_NATIVE_CHECKPOINT_BIN"] # The warm receiver uses the benchmark's execution profile inside its # isolated worker. Other cases retain their original sandbox contract. - sandbox = ("danger-full-access" if case in {"warm_preparation", "model_transition"} else - "workspace-write" if case == "joint" else "read-only") + sandbox = ("danger-full-access" if provider_transport or case in {"warm_preparation", "model_transition"} else + "workspace-write" if joint else "read-only") execution = Execution(context="resume", sandbox=sandbox) - with real_native_source(tmp_path / "native", binary, workspace, close_source=True, + with ExitStack() as receiver_transports, real_native_source(tmp_path / "native", binary, workspace, close_source=True, response_builder=response, sandbox=execution.sandbox) as (env, source_session, _, calls): env.update(LOOPX_REGISTRY=str(source_registry), LOOPX_RUNTIME_ROOT=str(source_runtime), LOOPX_GOAL_ID=source_goal, LOOPX_AGENT_ID=source_agent, @@ -479,7 +483,7 @@ def response(request, index, message): components = [*native["components"], {"id": "code", "source_path": str(candidate)}, {"id": "research", "source_path": str(research)}, {"id": "research_context", "source_path": str(context_export)}] - if case == "joint": + if joint: components.append({"id": "evaluation", "source_path": str(evaluator_unit)}) probe = effect_runtime_result("runtime.checkpoint_artifacts", {"action": "probe", "components": components}) unit = tmp_path / "point" @@ -493,7 +497,7 @@ def response(request, index, message): "provider_revision": source_context["snapshot"]["provider_revision"], "store_identity": source_context["snapshot"]["store_identity"]}}}, retry_safe=False) digest = captured["sha256"] - if case == "joint": + if joint: source_judge.state.judge.tokens[source_token]["submission_cooldown"] = 0 assert submit(source_url, source_token, 99).status_code == 200 judge_before = copy.deepcopy((source_judge.state.judge.tokens, @@ -539,6 +543,25 @@ def retained_source(): env.update(MODEL_NAME="fixture-next", REASONING_EFFORT="max") admission_options = {"destination_config": target_config} model_config_args = ["--destination-model-config", str(target_config)] + if provider_transport: + from tests.control_plane.codex_checkpoint_fixture import local_inference_provider + from benchmark.runtime.sforge_warm_start import admit_warm_session + + def receiver_response(body, index, message): + calls.append("/v1/responses") + return response(body, index + 1, message) + + receiver_url, _, _ = receiver_transports.enter_context(local_inference_provider(receiver_response)) + original = tomllib.loads(config.read_text()) + source_url = original["model_providers"][original["model_provider"]]["base_url"] + assert receiver_url != source_url + target_config = tmp_path / "destination-transport.toml" + target_config.write_text(config.read_text().replace(source_url, receiver_url)) + env["OPENAI_BASE_URL"] = receiver_url + admission_options = {"destination_config": target_config, "provider_base_url": receiver_url} + transport_declaration = {"root": str(unit), "sha256": digest, "config": str(config), + "destination_config": str(target_config), "provider_base_url": receiver_url, + "deadline_epoch": int(time.time()) + 600} lifecycle_registry = tmp_path / "lifecycle-only" / "registry.json" assert main(["--format", "json", "--registry", str(lifecycle_registry), "--runtime-root", str(tmp_path / "lifecycle-only/runtime"), "project", "register", @@ -641,12 +664,19 @@ def lose_binding_reply(self, *args): assert load_codex_cli_session(runtime, lineage={"goal_id": destination_goal, "agent_id": destination_agent}, session_scope="agent")["checkpoint_artifact_sha256"] == digest assert len(calls) == 1 - admitted_cli = subprocess.run([sys.executable, "-m", "loopx.hosts.codex_checkpoint", - "--root", str(unit), "--expected-sha256", digest, "--config-path", str(config), - "--sandbox", execution.sandbox, *model_config_args], - env=env, text=True, capture_output=True, timeout=40) - assert admitted_cli.returncode == 0, admitted_cli.stderr - admitted = json.loads(admitted_cli.stdout) + if provider_transport: + with pytest.raises(ValueError, match="only model and reasoning effort"): + admit_warm_session(env, {key: value for key, value in transport_declaration.items() + if key != "provider_base_url"}) + assert not Path(env["CODEX_HOME"]).exists() and len(calls) == 1 + admitted = admit_warm_session(env, transport_declaration) + else: + admitted_cli = subprocess.run([sys.executable, "-m", "loopx.hosts.codex_checkpoint", + "--root", str(unit), "--expected-sha256", digest, "--config-path", str(config), + "--sandbox", execution.sandbox, *model_config_args], + env=env, text=True, capture_output=True, timeout=40) + assert admitted_cli.returncode == 0, admitted_cli.stderr + admitted = json.loads(admitted_cli.stdout) assert admitted["session_binding_admitted"] and admitted["history_verified"] assert admitted["model_calls"] == 0 and not admitted["continuation_qualified"] assert admitted["session_id"] != source_session and len(calls) == 1 @@ -666,9 +696,12 @@ def forbidden_fork(*args, **kwargs): assert bound["checkpoint_artifact_sha256"] == digest # Transport reply loss leaves the atomic source-bound binding as the # original receipt. An exact CLI retry must also recover it. - retry_cli = subprocess.run(admitted_cli.args, env=env, text=True, capture_output=True, timeout=40) - assert retry_cli.returncode == 0, retry_cli.stderr - assert json.loads(retry_cli.stdout) == admitted + if provider_transport: + assert admit_warm_session(env, transport_declaration) == admitted + else: + retry_cli = subprocess.run(admitted_cli.args, env=env, text=True, capture_output=True, timeout=40) + assert retry_cli.returncode == 0, retry_cli.stderr + assert json.loads(retry_cli.stdout) == admitted with pytest.raises(ValueError, match="profile changed|destination model differs"): admit_native_checkpoint_session({**env, "REASONING_EFFORT": "low"}, unit, digest, config, sandbox=(execution).sandbox, timeout=30, **admission_options) @@ -749,7 +782,7 @@ def forbidden_fork(*args, **kwargs): assert json.loads(adopted_todo["todo"]["note"])["source_session"] == source_session guard = invoke([*prefix, "quota", "should-run", "--runtime-profile", "generic_cli", *binding]) assert guard["interaction_contract"]["agent_channel"]["delivery_allowed"] is True - if case == "joint": + if joint: control = tmp_path / "host-control" control.mkdir() branch_phase = control / "phase-deadline" @@ -785,7 +818,7 @@ def forbidden_fork(*args, **kwargs): response.raise_for_status() identifier = response.json()['submission_id'] for _ in range(100): - result = requests.get(url + '/api/v1/result/' + identifier, timeout=5).json() + result = requests.get(url + '/api/v1/result/' + identifier, params={'token': token}, timeout=5).json() if result['status'] == 'completed': break time.sleep(.01) assert result['status'] == 'completed' @@ -804,7 +837,7 @@ def forbidden_fork(*args, **kwargs): logs = tmp_path / "continuation" logs.mkdir() command = native_command(env, execution, "execute", logs, wake) - if case == "joint": + if joint: command = ["/bin/bash", "-c", shell + "exec " + shlex.join(command)] current_body = json.dumps(guard["interaction_contract"], ensure_ascii=False) result = subprocess.run(command, env=env, cwd=workspace, input=current_body + "\nContinue: verify the retained files.", @@ -848,7 +881,7 @@ def forbidden_fork(*args, **kwargs): assert "Future route outside this checkpoint" not in json.dumps(requests[2]) assert canonical_context({**env, "LOOPX_GOAL_ID": source_goal, "LOOPX_RUNTIME_ROOT": str(source_runtime)}, deadline=time.time() + 30) == later_context - if case == "joint": + if joint: assert "Captured public best 7; continued candidate 8" in json.dumps(requests[2]) history = http.get(branch_url + "/api/v1/history", params={"token": fresh["token"]}, timeout=5).json() assert history["best_score"] == 8 and history["best_round"] == "agent-3" diff --git a/benchmark/tests/test_sforge_warm_start.py b/benchmark/tests/test_sforge_warm_start.py index 6300e4cde5..7e56b190c4 100644 --- a/benchmark/tests/test_sforge_warm_start.py +++ b/benchmark/tests/test_sforge_warm_start.py @@ -68,13 +68,20 @@ def test_native_skills_backup_never_overwrites_an_earlier_attempt(tmp_path): @pytest.fixture -def point(tmp_path): +def point(tmp_path, request): workspace = tmp_path / "workspace" workspace.mkdir() (workspace / "candidate").write_text("retained candidate") (workspace / "candidate").chmod(0o600) config = tmp_path / "config.toml" - config.write_text('model="original"\nmodel_reasoning_effort="high"\n[features]\ngoals = false\n') + if getattr(request, "param", None) in {"gateway", "gateway-on"}: + from benchmark.runtime.codex import Execution, codex_settings + config.write_text(codex_settings(execution=Execution(context="resume", + context_delivery="on" if request.param == "gateway-on" else "off"), + workspace=workspace, model="original", effort="high", + base_url="http://gateway:4001/v1", wire_api="responses")) + else: + config.write_text('model="original"\nmodel_reasoning_effort="high"\n[features]\ngoals = false\n') context = tmp_path / "context.json" context.write_text(json.dumps({"snapshot": {"todos": [{"todo_id": "source-todo"}]}})) components = [{"id": "workspace", "source_path": str(workspace)}, @@ -208,6 +215,87 @@ async def exec(self, command, **kwargs): assert "destination_config" not in declaration +@pytest.mark.parametrize("point,context_delivery", [("gateway", "off"), ("gateway-on", "on")], indirect=["point"]) +@pytest.mark.parametrize("model_transition", [False, True]) +def test_fresh_gateway_stages_exact_first_worker_settings(point, monkeypatch, tmp_path, + model_transition, context_delivery): + from benchmark.runtime.codex import Execution, prepare_codex_home + + start, workspace = point + model, effort = "original", "high" + requested = start.config.read_bytes() + if model_transition: + target = tmp_path / "model.toml" + requested = requested.replace(b'"original"', b'"next"').replace(b'"high"', b'"max"') + target.write_bytes(requested) + start = replace(start, destination_config=target) + model, effort = "next", "max" + effective = requested.replace(b':4001/', b':4002/') + control = tmp_path / "control" + control.mkdir() + monkeypatch.setattr(warm, "_CONTROL", str(control)) + class Environment: + async def upload_dir(self, source, target): + shutil.copytree(source, target) + async def upload_file(self, source, target): + shutil.copyfile(source, target) + async def exec(self, *args, **kwargs): + return SimpleNamespace(return_code=0) + source_bytes = start.config.read_bytes() + asyncio.run(start.stage(Environment(), str(workspace), effective_config=effective, + provider_base_url="http://gateway:4002/v1")) + declaration = json.loads((control / "warm-start.json").read_bytes()) + destination = control / "restored-model-config.toml" + assert declaration["provider_base_url"] == "http://gateway:4002/v1" + assert destination.read_bytes() == effective + assert start.config.read_bytes() == source_bytes + assert declaration["deadline_epoch"] == start.deadline_epoch + # The native fork installs these destination bytes. The first ordinary + # worker must accept them and still refuse a later within-trial redirect. + home, skills = tmp_path / "receiver-home", tmp_path / "installed-skills" + home.mkdir() + skills.mkdir() + (home / "config.toml").write_bytes(effective) + args = dict(home=home, execution=Execution(context="resume", context_delivery=context_delivery), workspace=workspace, + model=model, effort=effort, base_url="http://gateway:4002/v1", + api_key="synthetic", wire_api="responses", skills=skills) + old_receiver = tmp_path / "unrebound-home" + old_receiver.mkdir() + (old_receiver / "config.toml").write_bytes(requested) + with pytest.raises(ValueError, match="settings changed within a trial"): + prepare_codex_home(**{**args, "home": old_receiver}) + assert (old_receiver / "config.toml").read_bytes() == requested + prepare_codex_home(**args) + with pytest.raises(ValueError, match="settings changed within a trial"): + prepare_codex_home(**{**args, "base_url": "http://gateway:4003/v1"}) + + +@pytest.mark.parametrize("point", ["gateway"], indirect=True) +@pytest.mark.parametrize("old,new", [(b'"original"', b'"undeclared"'), + (b'wire_api = "responses"', b'wire_api = "chat"'), + (b'web_search = "disabled"', b'web_search = "live"')]) +def test_rendered_route_cannot_change_declared_launch_before_staging(point, old, new): + start, workspace = point + effective = start.config.read_bytes().replace(b':4001/', b':4002/').replace(old, new) + class NoEffects: + def __getattr__(self, name): + pytest.fail("invalid configuration must reject before staging") + with pytest.raises(ValueError): + asyncio.run(start.stage(NoEffects(), str(workspace), effective_config=effective, + provider_base_url="http://gateway:4002/v1")) + + +@pytest.mark.parametrize("point", ["gateway"], indirect=True) +def test_unchanged_render_cannot_hide_a_different_declared_host_route(point): + start, workspace = point + class NoEffects: + def __getattr__(self, name): + pytest.fail("a mismatched Host route must reject before staging") + with pytest.raises(ValueError, match="current Host route"): + asyncio.run(start.stage(NoEffects(), str(workspace), + effective_config=start.config.read_bytes(), provider_base_url="http://gateway:4002/v1")) + + def test_normal_cadence_requires_accepted_current_replan_and_is_applied_once(tmp_path, monkeypatch): from benchmark.runtime import checkpoint_continuation, worker declaration = tmp_path / "warm.json" diff --git a/loopx/hosts/codex_checkpoint.py b/loopx/hosts/codex_checkpoint.py index 95fcbd2253..ccf93a2eea 100644 --- a/loopx/hosts/codex_checkpoint.py +++ b/loopx/hosts/codex_checkpoint.py @@ -282,17 +282,22 @@ def _native_thread_ids(home: Path) -> set[str]: def checkpoint_destination_config(config_path: Path, source_sha256: str, - destination_config: Path | None = None) -> tuple[bytes, dict[str, str] | None]: - """Verify sealed source settings and an explicit model/effort-only change. - - A destination is a new isolated receiver, never permission to change an - active session. Provider, tools, sandbox and all other configuration stay - semantically identical. Source bytes remain part of the sealed evidence. + destination_config: Path | None = None, *, + provider_base_url: str | None = None) -> tuple[bytes, dict[str, str] | None]: + """Verify sealed settings and a new receiver's explicit configuration. + + Model/effort may change as declared. A Host may separately bind its current + credential-gateway route; that changes only the existing provider's base URL, + never its protocol, credential contract, tools or permissions. The complete + source and destination config digests remain in the transition provenance. + This does not authorize changing an active session or selecting a provider. """ source = config_path.read_bytes() if hashlib.sha256(source).hexdigest() != source_sha256: raise ValueError("native checkpoint effective configuration differs") if destination_config is None: + if provider_base_url is not None: + raise ValueError("provider transport rebinding requires explicit destination settings") return source, None if (not destination_config.is_absolute() or destination_config.resolve(strict=True) != destination_config): @@ -303,9 +308,31 @@ def checkpoint_destination_config(config_path: Path, source_sha256: str, if any(not isinstance(config.get(key), str) or not config[key].strip() for config in (original, changed) for key in keys): raise ValueError("checkpoint model transition requires explicit model and reasoning effort") - if ({k: v for k, v in original.items() if k not in keys} != - {k: v for k, v in changed.items() if k not in keys}): - raise ValueError("checkpoint configuration transition may change only model and reasoning effort") + original_other = {k: v for k, v in original.items() if k not in keys} + changed_other = {k: v for k, v in changed.items() if k not in keys} + if provider_base_url is not None: + provider = original.get("model_provider") + before_profiles, after_profiles = original.get("model_providers"), changed.get("model_providers") + if (not isinstance(provider, str) or not isinstance(before_profiles, dict) + or not isinstance(after_profiles, dict)): + raise ValueError("checkpoint provider transport requires an existing custom provider") + before = before_profiles.get(provider, {}) + after = after_profiles.get(provider, {}) + if (not isinstance(provider_base_url, str) or not provider_base_url.strip() + or not isinstance(provider, str) or provider == "openai" + or changed.get("model_provider") != provider + or not isinstance(before, dict) or not isinstance(after, dict) + or not isinstance(before.get("base_url"), str) or not before["base_url"] + or after.get("base_url") != provider_base_url): + raise ValueError("checkpoint provider transport differs from the current Host route") + # Compare typed TOML values, not text substitutions or hostname guesses. + # All other provider entries and every non-route setting stay fixed. + for config, profile in ((original_other, before), (changed_other, after)): + config["model_providers"] = {**config["model_providers"], provider: { + key: value for key, value in profile.items() if key != "base_url"}} + if original_other != changed_other: + raise ValueError("checkpoint configuration transition may change only model and reasoning effort " + "or an explicitly bound provider transport route") return target, { "source_config_sha256": source_sha256, "destination_config_sha256": hashlib.sha256(target).hexdigest(), @@ -316,7 +343,8 @@ def checkpoint_destination_config(config_path: Path, source_sha256: str, def fork_native_checkpoint(env: dict[str, str], unit: Path, expected_sha256: str, config_path: Path, *, timeout: float, - destination_config: Path | None = None) -> dict[str, Any]: + destination_config: Path | None = None, + provider_base_url: str | None = None) -> dict[str, Any]: # The existing process lock serializes reconstruction in this isolated home. # The private attempt below describes one opaque native RPC, not authority # to admit a Goal, a work Turn, or a second fork after an uncertain response. @@ -324,6 +352,8 @@ def fork_native_checkpoint(env: dict[str, str], unit: Path, expected_sha256: str if not math.isfinite(timeout) or timeout <= 0: raise ValueError("native checkpoint reconstruction timeout must be finite and positive") + if provider_base_url is not None and provider_base_url != env.get("OPENAI_BASE_URL"): + raise ValueError("checkpoint provider transport differs from the current launch") deadline = time.monotonic() + timeout home = Path(env["CODEX_HOME"]) if not home.is_absolute() or home != home.resolve(): @@ -333,12 +363,14 @@ def fork_native_checkpoint(env: dict[str, str], unit: Path, expected_sha256: str if remaining <= 0: raise TimeoutError("native checkpoint reconstruction window closed") return _fork_native_checkpoint(env, unit, expected_sha256, config_path, timeout=remaining, - **({"destination_config": destination_config} if destination_config is not None else {})) + **({"destination_config": destination_config} if destination_config is not None else {}), + **({"provider_base_url": provider_base_url} if provider_base_url is not None else {})) def _fork_native_checkpoint(env: dict[str, str], unit: Path, expected_sha256: str, config_path: Path, *, timeout: float, - destination_config: Path | None = None) -> dict[str, Any]: + destination_config: Path | None = None, + provider_base_url: str | None = None) -> dict[str, Any]: """Reconstruct one native branch in a new, fixed-path isolated environment. No model starts and no LoopX binding is registered. Existing session/Goal @@ -374,7 +406,8 @@ def _fork_native_checkpoint(env: dict[str, str], unit: Path, expected_sha256: st if recovering and (not attempt_path.is_file() or attempt_path.resolve() != attempt_path): raise ValueError("native checkpoint destination home must be new or retain its original fork attempt") config, transition = checkpoint_destination_config( - config_path, native["config_sha256"], destination_config) + config_path, native["config_sha256"], destination_config, + **({"provider_base_url": provider_base_url} if provider_base_url is not None else {})) version = subprocess.check_output([env["CODEX_BIN"], "--version"], env=env, text=True, encoding="utf-8", timeout=max(.1, deadline - time.monotonic())).strip() if version != native["host_version"]: @@ -491,7 +524,8 @@ def request(method, params): def admit_native_checkpoint_session(env: dict[str, str], unit: Path, expected_sha256: str, config_path: Path, *, sandbox: str, timeout: float, - destination_config: Path | None = None) -> dict[str, Any]: + destination_config: Path | None = None, + provider_base_url: str | None = None) -> dict[str, Any]: """Explicit controller admission of a verified fork into a destination Goal. Existing native fork, Goal lifetime and session owners retain their rules. @@ -512,6 +546,8 @@ def admit_native_checkpoint_session(env: dict[str, str], unit: Path, expected_sh if not math.isfinite(timeout) or timeout <= 0: raise ValueError("checkpoint session admission timeout must be finite and positive") + if provider_base_url is not None and provider_base_url != env.get("OPENAI_BASE_URL"): + raise ValueError("checkpoint provider transport differs from the current launch") deadline = time.monotonic() + timeout effect_runtime_result("runtime.checkpoint_artifacts", { "action": "verify", "root": str(unit), "expected_sha256": expected_sha256, @@ -525,7 +561,8 @@ def admit_native_checkpoint_session(env: dict[str, str], unit: Path, expected_sh if tomllib.loads(config_path.read_text(encoding="utf-8")).get("features", {}).get("goals") is not False: raise ValueError("checkpoint session admission requires native automatic Goals explicitly disabled") config, transition = checkpoint_destination_config( - config_path, metadata["native_session"]["config_sha256"], destination_config) + config_path, metadata["native_session"]["config_sha256"], destination_config, + **({"provider_base_url": provider_base_url} if provider_base_url is not None else {})) target_profile = source_profile if transition is not None: if (env["MODEL_NAME"] != transition["destination_model"] or @@ -599,7 +636,8 @@ def admit_native_checkpoint_session(env: dict[str, str], unit: Path, expected_sh **({"model_transition": transition} if transition is not None else {})} result = fork_native_checkpoint(env, unit, expected_sha256, config_path, timeout=max(.1, deadline - time.monotonic()), - **({"destination_config": destination_config} if destination_config is not None else {})) + **({"destination_config": destination_config} if destination_config is not None else {}), + **({"provider_base_url": provider_base_url} if provider_base_url is not None else {})) receipt: dict[str, Any] = {} wake = CodexAgentSession(env, context="resume", sandbox=sandbox, receipt=receipt) # Hold the existing registry owner's fence through the session write. This