chore(benchmark): compose optional native collaboration parity

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,

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:
 

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",

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,

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:

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",