diff --git a/benchmark/edgebench/run.py b/benchmark/edgebench/run.py index 54cf6363a2..fd370c9eda 100644 --- a/benchmark/edgebench/run.py +++ b/benchmark/edgebench/run.py @@ -124,6 +124,8 @@ def main(argv=None): help="Opt-in short heartbeat context with same-invocation full captures") parser.add_argument("--context-delivery", choices=("off", "on"), help="Native Codex context delivery: on by default for heartbeat workers; pass off to disable") + parser.add_argument("--native-child-limit", type=int, + help="Explicit native child capacity and optional task delegation authorization, shared across study arms; omitted preserves existing behavior") 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", @@ -187,6 +189,9 @@ def main(argv=None): help="Host services run on Docker's default bridge (e.g. runner inside a VM); " "pin host.docker.internal there instead of Docker's host-gateway alias") args = parser.parse_args(argv) + if args.native_child_limit is not None and ( + args.native_child_limit < 1 or args.feedback != "native"): + parser.error("--native-child-limit requires a positive limit and native feedback") api_base_url = os.environ.get("OPENAI_BASE_URL") or None api_key = os.environ.get("OPENAI_API_KEY") if api_base_url else None if api_base_url and not api_key: @@ -288,6 +293,8 @@ def main(argv=None): archive_source(runner, pins[1], Path(directory) / "runner.tar", ("benchmark/runtime",)) task_file = args.tasks_dir / f"{args.task}.json" task = make_task_spec(task_file, load_benchmark(args.tasks_dir)) + if args.native_child_limit is not None and task.game_mode: + parser.error("--native-child-limit requires a non-game task with workspace instructions") research_sidecar = args.feedback == "native" and not task.game_mode if research_sidecar: from benchmark.runtime.research_artifacts import validate_research_scope, with_research_capture @@ -376,6 +383,7 @@ def main(argv=None): warm_start=warm_start, research_sidecar=research_sidecar, timeout_seconds=args.timeout, feedback_prompt=feedback_prompt, feedback=args.feedback, workspace_instructions=workspace_instructions, + native_child_limit=args.native_child_limit, task_objective=task_objective, task_entry=args.task_entry, turn_envelope=args.turn_envelope, diff --git a/benchmark/runtime/RUNTIME.md b/benchmark/runtime/RUNTIME.md index 017f507290..2b119d6968 100644 --- a/benchmark/runtime/RUNTIME.md +++ b/benchmark/runtime/RUNTIME.md @@ -30,6 +30,38 @@ existing trials pinned when comparing runner versions. ## Configure the native job +For a new EdgeBench study that explicitly permits host-native collaboration, +pass the same `--native-child-limit 3` to every arm. This gives each fresh Codex +profile the canonical `agents.max_concurrent_threads_per_session = 3` setting +(three open child sessions in addition to the parent), and the same optional delegation authorization +in the task input and workspace instructions. It neither requires delegation +nor grants separate child evaluation opportunities, time, accounts or access. +Only the parent submits candidates, and Stop still applies to the whole task. +The option requires a non-game task with native feedback. +Completed children retain a slot until closed; the shared instructions say to +close them when no longer needed. The adapter does not enable deeper delegation +or raise the native client's existing depth limit. + +LoopX arms additionally initialize their existing Goal spawn policy with the +same limit. Host-native children are ephemeral host sessions, not registered +LoopX Agents; this adds no peer scheduler, persistent Agent or delegation +binding. The existing native-child receipt and context owners remain unchanged. +The adapter reads back that Goal policy before later phases and restored work; +it does not change an existing Goal to make a mismatch pass. Repeated same-session +entry preserves the fixed configuration. Restore with the same declared limit; +a different or omitted limit cannot silently replace the sealed native config. + +Omitting the option preserves prior configuration, instructions and receipt +fields. The worker environment explicitly emits an empty child-limit value for +the omitted option so a container's ambient value cannot enable it; this is the +only intentional default environment/command serialization change. Do not add +the option to a running study. Record the explicit authorization as +a new protocol factor: arms in the new study share it, while comparisons to an +older study cannot attribute changes solely to context delivery or another fix. +Roll back by starting a separate study without the option, not by editing a +continuing session. Capacity/configuration acceptance is not evidence that a +model delegated useful work or that its score improved. + Use this agent in the benchmark's existing job config, retaining its dataset and environment settings: diff --git a/benchmark/runtime/codex.py b/benchmark/runtime/codex.py index 203c560678..3464c2407d 100644 --- a/benchmark/runtime/codex.py +++ b/benchmark/runtime/codex.py @@ -29,8 +29,13 @@ class Execution: task_entry: str | None = None turn_envelope: bool = False context_delivery: str | None = None + native_child_limit: int | None = None def __post_init__(self) -> None: + if self.native_child_limit is not None and ( + type(self.native_child_limit) is not int or self.native_child_limit < 1 + ): + raise ValueError("native_child_limit must be a positive integer or None") if self.context is None: object.__setattr__(self, "context", "resume" if self.mode in {"heartbeat", "turn"} else "fresh") if self.context_delivery is None: @@ -81,6 +86,24 @@ def turn_envelope_receipt(*, explicit: bool) -> dict: return {} +def native_child_instructions(limit: int | None) -> str: + """One explicit task authorization shared by every opted-in study arm.""" + if limit is None: + return "" + Execution(mode="plain", native_child_limit=limit) + return ( + "\n\n## Optional native collaboration\n\n" + f"You are authorized to use up to {limit} open host-native child sessions " + "for independent work on this task. Delegation is optional; do not create work " + "merely to fill capacity. Continue useful parent work and verify child results. " + "Completed children still occupy slots; close them when their work is no longer " + "needed to free capacity. " + "Children share the original task scope, permissions, total time allowance and " + "evaluation opportunities. Only the parent submits candidates to the evaluator. " + "This does not authorize registering persistent LoopX Agents, expanding access, " + "starting another experiment, or extending a deadline. Preserve explicit Stop " + "and close child work before the parent finishes.\n" + ) def validate_model_catalog(model_catalog: Path, model: str) -> dict: """Admit only catalog entries that keep Codex's function-tool surface. @@ -164,6 +187,8 @@ def codex_settings( "[features]", f"goals = {str(execution.native_goal).lower()}", "unified_exec = true", + *(["[agents]", f"max_concurrent_threads_per_session = {execution.native_child_limit}"] + if execution.native_child_limit is not None else []), "[memories]", "generate_memories = false", "use_memories = false", diff --git a/benchmark/runtime/harbor.py b/benchmark/runtime/harbor.py index 6ecd140fee..8d2e353984 100644 --- a/benchmark/runtime/harbor.py +++ b/benchmark/runtime/harbor.py @@ -74,6 +74,7 @@ def __init__( planning_timeout_sec=None, turn_envelope=False, context_delivery=None, + native_child_limit=None, provider_request_admission=False, native_goal_continue=False, task_objective=None, @@ -100,6 +101,7 @@ def __init__( task_entry, turn_envelope, context_delivery, + native_child_limit, ) if type(provider_request_admission) is not bool: raise ValueError("provider_request_admission must be a boolean opt-in") @@ -190,6 +192,8 @@ def __init__( def _turn_envelope_receipt(self) -> dict: return {**turn_envelope_receipt(explicit=self.execution.turn_envelope), + **({"native_child_limit": self.execution.native_child_limit} + if self.execution.native_child_limit is not None else {}), **({"context_delivery": "on"} if self.execution.context_delivery != "off" else {})} def _planning_receipt(self) -> dict: @@ -478,8 +482,14 @@ async def _prepare_phase( ) if pending.return_code == 0: raise RuntimeError("Resolve the pending Turn before entering another task phase") + registry_exists = await self._registry_exists(environment) + if registry_exists and self.execution.native_child_limit is not None: + retained = await self._loopx( + environment, ["configure-goal", "--goal-id", self.goal_id], cwd=cwd, + ) + self._verify_native_child_configuration(retained.get("after") or retained.get("before") or {}) await self._write_task_document(environment, instruction) - if not await self._registry_exists(environment): + if not registry_exists: await self._loopx( environment, [ @@ -500,6 +510,8 @@ async def _prepare_phase( "--write-scope", "**", "--no-global-sync", + *(["--spawn-allowed", "--max-children", str(self.execution.native_child_limit)] + if self.execution.native_child_limit is not None else []), ], cwd=cwd, ) @@ -554,6 +566,7 @@ async def _verify_phase_configuration(self, environment: BaseEnvironment, *, cwd environment, ["configure-goal", "--goal-id", self.goal_id], cwd=cwd, ) configured_state = cadence.get("after") or cadence.get("before") or {} + self._verify_native_child_configuration(configured_state) configured = configured_state.get("execution_profile", {}).get(key) if configured != expected: raise RuntimeError( @@ -568,6 +581,15 @@ async def _verify_phase_configuration(self, environment: BaseEnvironment, *, cwd f"stride strategy readback mismatch: expected {self.stride_strategy or 'off'}, got {strategy!r}" ) + def _verify_native_child_configuration(self, configured_state: dict) -> None: + if self.execution.native_child_limit is None: + return + orchestration = configured_state.get("orchestration") or {} + if (orchestration.get("mode") != "multi_subagent" + or orchestration.get("spawn_allowed") is not True + or orchestration.get("max_children") != self.execution.native_child_limit): + raise RuntimeError("Native child authorization differs from the existing Goal; use a new trial") + async def _seed_phase(self, environment: BaseEnvironment, *, cwd: str) -> None: text = ( f"[P0] Complete the task in {self._task_document}. Inspect the workspace, " @@ -641,6 +663,11 @@ 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": self.execution.context_delivery, + # A trial's explicit None must not inherit a container opt-in. + "LOOPX_NATIVE_CHILD_LIMIT": ( + str(self.execution.native_child_limit) + if self.execution.native_child_limit is not None 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, diff --git a/benchmark/runtime/sforge.py b/benchmark/runtime/sforge.py index c742e69b26..66b94a75f3 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, codex_settings, prepare_codex_home +from .codex import DEFAULT_REPLAN_AFTER_TURNS, Execution, codex_settings, native_child_instructions, 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, @@ -93,6 +93,7 @@ def __init__(self, config, *, profile: str, cwd: str, task_entry: str | None = None, turn_envelope: bool = False, context_delivery: str | None = None, + native_child_limit: int | None = None, provider_request_admission: bool = False, native_goal_continue: bool = False, replan_after_turns: int | None = None, @@ -110,6 +111,13 @@ def __init__(self, config, *, profile: str, cwd: str, ): raise ValueError("turn_envelope requires a heartbeat worker") self.turn_envelope = turn_envelope + self.native_child_limit = Execution(mode="plain", native_child_limit=native_child_limit).native_child_limit + if native_child_limit is not None: + if not feedback_prompt or not workspace_instructions: + raise ValueError("Native child authorization requires the explicit task prompt and workspace instructions") + authorization = native_child_instructions(native_child_limit) + feedback_prompt = authorization + feedback_prompt + workspace_instructions = authorization + workspace_instructions self.context_delivery = Execution( mode="heartbeat" if profile.startswith("heartbeat-") else "plain", context_delivery=context_delivery, @@ -305,7 +313,7 @@ def _install_worker(self, backend, handle, log_dir, logger): shutil.copyfile(os.environ["CODEX_AUTH_JSON_PATH"], home / "auth.json") (home / "auth.json").chmod(0o600) prepare_codex_home( - home, execution=Execution(mode="plain"), workspace=Path(self.cwd), + home, execution=Execution(mode="plain", native_child_limit=self.native_child_limit), workspace=Path(self.cwd), model=self._config.agent_model, effort=effort, base_url=runtime_base_url or "", api_key=runtime_api_key or "", wire_api="responses", skills=None, @@ -336,6 +344,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, + native_child_limit=self.native_child_limit, **({"provider_request_admission": True} if self.provider_request_admission else {}), native_goal_continue=self.native_goal_continue, replan_after_turns=self.replan_after_turns, @@ -428,6 +437,8 @@ def _install_worker(self, backend, handle, log_dir, logger): **({"planning_timeout_seconds": None} if self.task_entry == "loopx-planned" else {}), "stop_hook": self.profile == "official", + **({"native_child_limit": self.native_child_limit} + if self.native_child_limit is not None else {}), **(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 {}), @@ -482,8 +493,11 @@ def format_run_cmd(self, prompt_path, *, model=None, cwd="", internet=True, resu # The sealed Goal, task graph, review counters and strategy # are the continuing state. Do not bootstrap, seed another # task, or reset an existing configuration during restore. + if self.native_child_limit is not None: + asyncio.run(self.runtime._verify_phase_configuration(self.environment, cwd=self.cwd)) asyncio.run(self.runtime._write_task_document(self.environment, result.stdout)) - asyncio.run(self.runtime._verify_phase_configuration(self.environment, cwd=self.cwd)) + if self.native_child_limit is None: + asyncio.run(self.runtime._verify_phase_configuration(self.environment, cwd=self.cwd)) else: asyncio.run(self.runtime._prepare_phase(self.environment, result.stdout, cwd=self.cwd)) if self.profile == "heartbeat-explore" and self.warm_start is None: diff --git a/benchmark/runtime/worker.py b/benchmark/runtime/worker.py index f7673de2ef..d6239c5e0d 100644 --- a/benchmark/runtime/worker.py +++ b/benchmark/runtime/worker.py @@ -505,6 +505,8 @@ def run_once(env: dict[str, str]) -> dict: task_entry=env.get("LOOPX_TASK_ENTRY"), turn_envelope=env.get("LOOPX_TURN_ENVELOPE", "0") == "1", context_delivery=env.get("LOOPX_CONTEXT_DELIVERY"), + native_child_limit=(int(env["LOOPX_NATIVE_CHILD_LIMIT"]) + if env.get("LOOPX_NATIVE_CHILD_LIMIT") else None), ) request_admission = False if env.get("LOOPX_PROVIDER_REQUEST_ADMISSION", "off") != "off": @@ -551,6 +553,8 @@ 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 {}), + **({"native_child_limit": execution.native_child_limit} + if execution.native_child_limit is not None else {}), **({"provider_request_admission": "on"} if request_admission else {}), "stage": stage, "home_scope": "trial", diff --git a/benchmark/tests/test_edgebench_prompts.py b/benchmark/tests/test_edgebench_prompts.py index ad753c6d0b..62b56db096 100644 --- a/benchmark/tests/test_edgebench_prompts.py +++ b/benchmark/tests/test_edgebench_prompts.py @@ -1,10 +1,12 @@ from pathlib import Path from types import SimpleNamespace +import asyncio import json import os import subprocess import sys import tempfile +import time import pytest @@ -194,8 +196,9 @@ def test_native_goal_transport_rejects_objective_outside_task_prompt(objective, @pytest.mark.parametrize("profile", ["heartbeat-resume", "heartbeat-explore"]) +@pytest.mark.parametrize("native_child_limit", [None, 3]) def test_native_research_objective_and_full_planning_input_survive_real_goal_bootstrap( - tmp_path, monkeypatch, profile): + tmp_path, monkeypatch, profile, native_child_limit): pytest.importorskip("sforge") pytest.importorskip("harbor") from sforge.harness.config import SForgeConfig @@ -243,13 +246,34 @@ async def upload_file(self, source, target): prompt = native_task_prompt(query, ["solution.py"], **options) worker = SForgeWorker(SForgeConfig(agent_model="fixture", agent_effort="xhigh"), profile=profile, cwd=str(project), feedback_prompt=prompt, workspace_instructions=policy, - task_objective=query) + task_objective=query, native_child_limit=native_child_limit) + from benchmark.runtime.codex import native_child_instructions + prompt = native_child_instructions(native_child_limit) + prompt + policy = native_child_instructions(native_child_limit) + policy worker.install_stop_hook(None, None, tmp_path, None) worker.environment = Environment() remote = control / "entry.md" try: command = worker.format_run_cmd(str(remote), internet=False) registry = json.loads(Path(harbor._REGISTRY).read_text()) + spawn = registry["goals"][0]["spawn_policy"] + assert spawn["allowed"] is (native_child_limit is not None) + assert spawn["mode"] == ("multi_subagent" if native_child_limit else "default") + assert spawn["max_children"] == 3 + assert registry["goals"][0]["coordination"]["registered_agents"] == ["benchmark-agent"] + delegation = asyncio.run(worker.runtime._loopx(worker.environment, [ + "agent-context", "--goal-id", "benchmark-goal", "--agent-id", "benchmark-agent", + "--phase", "before_plan", + ], cwd=str(project))) + assert delegation["read_only"] is True + assert delegation["host_receipts_observed"] is False + context = delegation["agent_context"] + if native_child_limit is None: + assert context is None + else: + contribution = next(c for c in context["contributions"] if c["capability_id"] == "multi_subagent") + assert contribution["facts"]["max_children"] == 3 + assert contribution["facts"]["capacity_contract"]["live_availability"] == "not_observed" state = Path(registry["goals"][0]["state_file"]) if not state.is_absolute(): state = project / state @@ -280,6 +304,30 @@ async def upload_file(self, source, target): assert core["objective_source"] == "active_state" and core["acceptance_contract"] is None assert worker.format_run_cmd(str(remote), internet=False, resume=True) == command assert (project / "AGENTS.md").read_text().count(policy) == 1 + if native_child_limit is not None: + before_registry = Path(harbor._REGISTRY).read_bytes() + before_state = state.read_bytes() + before_task = Path(harbor._TASK_DOC).read_bytes() + asyncio.run(worker.runtime._verify_phase_configuration(worker.environment, cwd=str(project))) + assert Path(harbor._REGISTRY).read_bytes() == before_registry + assert state.read_bytes() == before_state + # A receiver may not repair an incompatible retained Goal implicitly. + from dataclasses import replace + worker.runtime.execution = replace(worker.runtime.execution, native_child_limit=2) + with pytest.raises(RuntimeError, match="Native child authorization differs"): + asyncio.run(worker.runtime._prepare_phase(worker.environment, "Mismatched receiver input", cwd=str(project))) + assert Path(harbor._REGISTRY).read_bytes() == before_registry + assert state.read_bytes() == before_state + assert Path(harbor._TASK_DOC).read_bytes() == before_task + # The same read-before-write requirement applies to the ordinary + # warm receiver entrypoint after workspace restoration. + worker.prepared = False + worker.warm_start = SimpleNamespace(deadline_epoch=time.time() + 300) + with pytest.raises(RuntimeError, match="Native child authorization differs"): + worker.format_run_cmd(str(remote), internet=False) + assert Path(harbor._REGISTRY).read_bytes() == before_registry + assert state.read_bytes() == before_state + assert Path(harbor._TASK_DOC).read_bytes() == before_task finally: subprocess.run([sys.executable, "-c", "from loopx.control_plane.effect_runtime import " "effect_runtime_result; effect_runtime_result('runtime.shutdown',{},retry_safe=False)"], diff --git a/benchmark/tests/test_native_child_parity.py b/benchmark/tests/test_native_child_parity.py new file mode 100644 index 0000000000..46ccb85cda --- /dev/null +++ b/benchmark/tests/test_native_child_parity.py @@ -0,0 +1,103 @@ +"""Matched optional host-native delegation, with immutable continuing profiles.""" +from dataclasses import replace +import tomllib + +import pytest + +from benchmark.runtime.codex import Execution, codex_settings, native_child_instructions, prepare_codex_home + + +@pytest.mark.parametrize("value", [True, False, 0, -1, 1.5, "3"]) +def test_child_limit_rejects_nonpositive_or_untyped_inputs(value): + with pytest.raises(ValueError, match="native_child_limit"): + Execution(native_child_limit=value) + + +@pytest.mark.parametrize("mode", ["plain", "native-goal", "heartbeat"]) +def test_all_hosts_get_identical_explicit_capacity_and_default_stays_absent(tmp_path, mode): + kwargs = dict(workspace=tmp_path, model="fixture", effort="xhigh", base_url="", wire_api="responses") + legacy = codex_settings(execution=Execution(mode=mode), **kwargs) + assert "[agents]" not in legacy + assert legacy == codex_settings(execution=Execution(mode=mode, native_child_limit=None), **kwargs) + configured = codex_settings(execution=Execution(mode=mode, native_child_limit=3), **kwargs) + data = tomllib.loads(configured) + assert data["agents"] == {"max_concurrent_threads_per_session": 3} + assert configured.replace("[agents]\nmax_concurrent_threads_per_session = 3\n", "") == legacy + + +def test_same_session_and_restored_home_cannot_drift_capacity(tmp_path): + home, skills = tmp_path / "home", tmp_path / "skills" + skills.mkdir() + execution = Execution(native_child_limit=3) + args = dict(home=home, execution=execution, workspace=tmp_path, model="fixture", + effort="xhigh", base_url="http://127.0.0.1:1/v1", api_key="fixture", + wire_api="responses", skills=skills) + prepare_codex_home(**args) + session = home / "sessions" / "original.jsonl" + session.parent.mkdir() + session.write_text('{"original_session":true}\n') + original = {p.relative_to(home): p.read_bytes() for p in [home / "config.toml", session]} + prepare_codex_home(**args) + assert {p: (home / p).read_bytes() for p in original} == original + # These are also the checks run by the receiving worker after sealed restore. + for limit in [None, 2, 4]: + with pytest.raises(ValueError, match="settings changed within a trial"): + prepare_codex_home(**(args | {"execution": replace(execution, native_child_limit=limit)})) + assert {p: (home / p).read_bytes() for p in original} == original + + +@pytest.mark.parametrize("profile", ["official", "native-goal", "heartbeat-explore"]) +def test_same_task_authorization_and_no_extra_persistent_agents(tmp_path, monkeypatch, profile): + pytest.importorskip("sforge") + pytest.importorskip("harbor") + from sforge.harness.config import SForgeConfig + from benchmark.runtime.sforge import SForgeWorker + monkeypatch.setenv("CODEX_AUTH_JSON_PATH", "/synthetic-credential") + args = dict(config=SForgeConfig(agent_model="fixture", agent_effort="xhigh"), + profile=profile, cwd=str(tmp_path), feedback_prompt="Original workspace policy\nOriginal task", + workspace_instructions="Original workspace policy") + legacy = SForgeWorker(**args) + assert legacy.feedback_prompt == "Original workspace policy\nOriginal task" + assert legacy.workspace_instructions == "Original workspace policy" + enabled = SForgeWorker(**args, native_child_limit=3) + shared = native_child_instructions(3) + assert enabled.feedback_prompt == shared + "Original workspace policy\nOriginal task" + assert enabled.workspace_instructions == shared + "Original workspace policy" + assert "Only the parent submits" in shared + assert "does not authorize registering persistent LoopX Agents" in shared + assert "optional" in shared and "Stop" in shared + with pytest.raises(ValueError, match="explicit task prompt"): + SForgeWorker(**(args | {"feedback_prompt": None}), native_child_limit=3) + # The existing workspace-policy owner also guards direct API callers; + # native-only authorization cannot be injected into a restricted wrapper. + with pytest.raises(ValueError, match="native task wrapper"): + SForgeWorker(**(args | {"feedback": "blind"}), native_child_limit=3) + + +def test_worker_environment_propagates_only_explicit_capacity(tmp_path, monkeypatch): + pytest.importorskip("harbor") + from benchmark.runtime.harbor import BenchmarkCodex + monkeypatch.setenv("LOOPX_NATIVE_CHILD_LIMIT", "99") + for mode in ["heartbeat", "native-goal"]: + agent = BenchmarkCodex(logs_dir=tmp_path / mode, model_name="fixture", execution_mode=mode, + native_child_limit=3) + agent._task_path = "/usr/bin:/bin" + agent._profile_env = lambda: {} + env = agent._worker_env(cwd=str(tmp_path)) + assert env["LOOPX_NATIVE_CHILD_LIMIT"] == "3" + agent.execution = replace(agent.execution, native_child_limit=None) + assert agent._worker_env(cwd=str(tmp_path))["LOOPX_NATIVE_CHILD_LIMIT"] == "" + + +@pytest.mark.parametrize("limit,feedback", [("0", "native"), ("-1", "native"), ("3", "blind")]) +def test_cli_rejects_invalid_collaboration_before_creating_a_trial(tmp_path, limit, feedback): + pytest.importorskip("sforge") + pytest.importorskip("harbor") + from benchmark.edgebench.run import main + with pytest.raises(SystemExit) as rejected: + main(["--task", "fixture", "--tasks-dir", str(tmp_path), "--log-dir", str(tmp_path), + "--run-id", "unadmitted", "--worker", "heartbeat-explore", "--model", "fixture", + "--effort", "high", "--judge-url", "http://localhost:1", "--feedback", feedback, + "--native-child-limit", limit]) + assert rejected.value.code == 2 + assert not list(tmp_path.iterdir()) diff --git a/benchmark/tests/test_sforge_warm_start.py b/benchmark/tests/test_sforge_warm_start.py index 527834bfd5..04cdce6a68 100644 --- a/benchmark/tests/test_sforge_warm_start.py +++ b/benchmark/tests/test_sforge_warm_start.py @@ -76,10 +76,11 @@ def point(tmp_path, request, monkeypatch): (workspace / "candidate").write_text("retained candidate") (workspace / "candidate").chmod(0o600) config = tmp_path / "config.toml" - if getattr(request, "param", None) in {"gateway", "gateway-on"}: + if getattr(request, "param", None) in {"gateway", "gateway-on", "gateway-children"}: 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"), + context_delivery="on" if request.param == "gateway-on" else "off", + native_child_limit=3 if request.param == "gateway-children" else None), workspace=workspace, model="original", effort="high", base_url="http://gateway:4001/v1", wire_api="responses")) else: @@ -231,10 +232,12 @@ 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("point,context_delivery,native_child_limit", [ + ("gateway", "off", None), ("gateway-on", "on", None), ("gateway-children", "off", 3), +], 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): + model_transition, context_delivery, native_child_limit): from benchmark.runtime.codex import Execution, prepare_codex_home start, workspace = point @@ -272,7 +275,8 @@ async def exec(self, *args, **kwargs): home.mkdir() skills.mkdir() (home / "config.toml").write_bytes(effective) - args = dict(home=home, execution=Execution(context="resume", context_delivery=context_delivery), workspace=workspace, + args = dict(home=home, execution=Execution(context="resume", context_delivery=context_delivery, + native_child_limit=native_child_limit), 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" @@ -282,10 +286,28 @@ async def exec(self, *args, **kwargs): prepare_codex_home(**{**args, "home": old_receiver}) assert (old_receiver / "config.toml").read_bytes() == requested prepare_codex_home(**args) + if native_child_limit is not None: + for changed in [None, 2]: + with pytest.raises(ValueError, match="settings changed within a trial"): + prepare_codex_home(**{**args, "execution": replace(args["execution"], native_child_limit=changed)}) + assert (home / "config.toml").read_bytes() == effective 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-children"], indirect=True) +def test_warm_render_cannot_change_native_child_authorization(point): + start, workspace = point + effective = start.config.read_bytes().replace(b':4001/', b':4002/').replace( + b'max_concurrent_threads_per_session = 3', b'max_concurrent_threads_per_session = 2') + class NoEffects: + def __getattr__(self, name): + pytest.fail("capacity mismatch must reject before restoring workspace or authority") + 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) @pytest.mark.parametrize("old,new", [(b'"original"', b'"undeclared"'), (b'wire_api = "responses"', b'wire_api = "chat"'), diff --git a/benchmark/tests/test_task_entry.py b/benchmark/tests/test_task_entry.py index 4ed96b4f73..42f11fa6d5 100644 --- a/benchmark/tests/test_task_entry.py +++ b/benchmark/tests/test_task_entry.py @@ -341,11 +341,15 @@ def test_real_planning_rejects_todos_without_agent_requirements(planning_env): assert not packet["goal_acceptance"]["planning_requirements"]["declaration_current"] -def test_stopped_goal_cannot_launch_first_planner_or_fabricate_requirements(planning_env): +@pytest.mark.parametrize("native_child_limit", [None, 3]) +def test_stopped_goal_cannot_launch_first_planner_or_fabricate_requirements(planning_env, native_child_limit): from loopx.control_plane.goals.first_party_host_admission import FirstPartyHostRuntimeRejected from benchmark.runtime.planning import task_plan_packet from benchmark.runtime.worker import loopx_command + if native_child_limit is not None: + planning_env["LOOPX_NATIVE_CHILD_LIMIT"] = str(native_child_limit) + command = loopx_command(planning_env) + ["goal-lifecycle", "--goal-id", "planning-goal", "--operation", "stop"] preview = json.loads(subprocess.run(command, env=planning_env, capture_output=True, text=True, check=True).stdout)