Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 10 additions & 1 deletion benchmark/edgebench/run.py
Original file line number Diff line number Diff line change
Expand Up @@ -126,6 +126,8 @@ def main(argv=None):
help="Native Codex context delivery: on by default for heartbeat workers; pass off to disable")
parser.add_argument("--execution-observation", choices=("local_probe",),
help="Optional local execution probe for heartbeat workers")
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",
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -286,6 +291,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
Expand Down Expand Up @@ -357,7 +364,8 @@ def main(argv=None):
args.restore_goal, args.restore_agent, args.restore_source_todo,
int(time.time()) + args.timeout, args.replan_after_turns or 6,
tuple(args.restore_research_component), args.restore_fresh_replan,
destination_config=args.restore_model_config, research_sidecar=research_sidecar)
destination_config=args.restore_model_config, research_sidecar=research_sidecar,
native_child_limit=args.native_child_limit)
metadata = candidate.verify(task.cwd)
if args.restore_model_config is not None:
from loopx.hosts.codex_checkpoint import checkpoint_destination_config
Expand All @@ -374,6 +382,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,
Comment thread
loopx-agent marked this conversation as resolved.
Resolved
task_objective=task_objective,
task_entry=args.task_entry,
turn_envelope=args.turn_envelope,
Expand Down
32 changes: 32 additions & 0 deletions benchmark/runtime/RUNTIME.md
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,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. Warm restore compares the requested
limit, including omission, with the sealed native config before native activation.

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:

Expand Down
25 changes: 25 additions & 0 deletions benchmark/runtime/codex.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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",
Expand Down
33 changes: 32 additions & 1 deletion benchmark/runtime/harbor.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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")
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -489,8 +493,18 @@ async def _prepare_phase(
)
if pending.return_code == 0:
raise RuntimeError("Resolve the pending Turn before entering another task phase")
# Check explicitly requested capacity before replacing any task document.
registry_exists = (await self._registry_exists(environment)
if self.execution.native_child_limit is not None else None)
if registry_exists:
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 registry_exists is None:
registry_exists = await self._registry_exists(environment)
if not registry_exists:
await self._loopx(
environment,
[
Expand All @@ -511,6 +525,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,
)
Expand Down Expand Up @@ -565,6 +581,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(
Expand All @@ -579,6 +596,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, "
Expand Down Expand Up @@ -652,6 +678,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,
Expand Down
23 changes: 20 additions & 3 deletions benchmark/runtime/sforge.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -95,6 +95,7 @@ def __init__(self, config, *, profile: str, cwd: str,
turn_envelope: bool = False,
context_delivery: str | None = None,
execution_observation: 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,
Expand All @@ -112,6 +113,9 @@ 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 self.native_child_limit is not None and (not feedback_prompt or not workspace_instructions):
raise ValueError("Native child authorization requires the explicit task prompt and workspace instructions")
self.context_delivery = Execution(
mode="heartbeat" if profile.startswith("heartbeat-") else "plain",
context_delivery=context_delivery,
Expand Down Expand Up @@ -198,6 +202,8 @@ def __init__(self, config, *, profile: str, cwd: str,
raise ValueError("Managed research storage requires matching warm receiver selection")
if not profile.startswith("heartbeat-") or feedback != "native" or self.task_entry != "loopx-planned":
raise ValueError("Warm checkpoint continuation requires a planned native heartbeat worker")
if getattr(warm_start, "native_child_limit", None) != self.native_child_limit:
raise ValueError("Worker native child limit differs from the warm start request")
warm_start.verify(cwd)
self.profile, self.cwd = profile, cwd
if feedback not in FEEDBACK_MODES:
Expand All @@ -220,6 +226,11 @@ def __init__(self, config, *, profile: str, cwd: str,
or task_objective not in feedback_prompt
):
raise ValueError("Task objective must occur in the native task prompt")
if self.native_child_limit is not None:
assert feedback_prompt is not None and workspace_instructions is not None
authorization = native_child_instructions(self.native_child_limit)
feedback_prompt = authorization + feedback_prompt
workspace_instructions = authorization + workspace_instructions
self.feedback = feedback
self.feedback_prompt = feedback_prompt
self.workspace_instructions = workspace_instructions
Expand Down Expand Up @@ -310,7 +321,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,
Expand Down Expand Up @@ -343,6 +354,7 @@ def _install_worker(self, backend, handle, log_dir, logger):
context_delivery=self.context_delivery,
**({"execution_observation": self.execution_observation}
if self.execution_observation is not None else {}),
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,
Expand Down Expand Up @@ -435,6 +447,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 {}),
Expand Down Expand Up @@ -491,8 +505,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:
Expand Down
Loading
Loading