From a622482144dc005e077cd4bdfd3923c11e2dece9 Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Fri, 18 Sep 2026 02:39:11 +0800 Subject: [PATCH 1/7] feat(todo): expose caller-owned task planning checkpoint Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- loopx/cli_commands/todo.py | 17 +- .../cli_commands/todo_argument_validation.py | 11 +- loopx/cli_commands/todo_registration.py | 2 + loopx/control_plane/goals/start_contract.py | 85 +++++----- .../goals/start_goal_todo_delta.py | 6 +- loopx/control_plane/goals/task_planning.py | 149 ++++++++++++++++++ loopx/slash_command_install.py | 1 + tests/control_plane/test_task_planning.py | 123 +++++++++++++++ 8 files changed, 351 insertions(+), 43 deletions(-) create mode 100644 loopx/control_plane/goals/task_planning.py create mode 100644 tests/control_plane/test_task_planning.py diff --git a/loopx/cli_commands/todo.py b/loopx/cli_commands/todo.py index 322b66ecc1..a0453ccf3b 100644 --- a/loopx/cli_commands/todo.py +++ b/loopx/cli_commands/todo.py @@ -30,6 +30,10 @@ build_todo_suggestion_prompt_packet, render_todo_suggestion_prompt_markdown, ) +from ..control_plane.goals.task_planning import ( + build_task_planning_packet, + render_task_planning_packet, +) from ..todos import ( add_goal_todo, archive_completed_todos, @@ -50,6 +54,7 @@ validate_todo_list_options, validate_todo_project_markdown_options, validate_todo_suggest_options, + validate_todo_plan_options, validate_todo_supersede_options, validate_todo_update_options, ) @@ -195,6 +200,9 @@ def handle_todo_command( post_writeback_projection_builder: PostWritebackProjectionBuilder | None = None, ) -> int: renderer = ( + render_task_planning_packet + if args.todo_command == "plan" + else render_todo_suggestion_prompt_markdown if args.todo_command == "suggest" else render_todo_markdown @@ -208,7 +216,14 @@ def handle_todo_command( ) validate_shared_todo_options(args) validate_capability_gap_options(args) - if args.todo_command == "list": + if args.todo_command == "plan": + validate_todo_plan_options(args) + payload = build_task_planning_packet( + registry_path=registry_path, runtime_root_arg=runtime_root_arg, + goal_id=args.goal_id, agent_id=args.agent_id, text=args.text, + project=Path(args.project).expanduser() if args.project else None, + ) + elif args.todo_command == "list": validate_todo_list_options(args) payload = list_goal_todos( registry_path=registry_path, diff --git a/loopx/cli_commands/todo_argument_validation.py b/loopx/cli_commands/todo_argument_validation.py index b980ecc073..6753f36b88 100644 --- a/loopx/cli_commands/todo_argument_validation.py +++ b/loopx/cli_commands/todo_argument_validation.py @@ -303,6 +303,15 @@ def validate_todo_list_options(args: argparse.Namespace) -> None: ) +def validate_todo_plan_options(args: argparse.Namespace) -> None: + _validate_todo_option_subset( + args, {"text", "agent_id"}, + "todo plan only accepts --goal-id, --agent-id, --text, --project and --format; unsupported: ", + ) + if not args.text or not args.agent_id: + raise ValueError("todo plan requires --text and a registered --agent-id") + + def validate_todo_project_markdown_options(args: argparse.Namespace) -> None: if not getattr(args, "provider_revision", None): raise ValueError("todo project-markdown requires --provider-revision") @@ -551,7 +560,7 @@ def validate_shared_todo_options(args: argparse.Namespace) -> None: "--authority-reason is supported only by todo update/complete/supersede" ) if ( - args.todo_command not in {"suggest", "capture-followups"} + args.todo_command not in {"suggest", "plan", "capture-followups"} and args.agent_id and not agent_id_allowed_for_user_authoring and not agent_id_allowed_for_read diff --git a/loopx/cli_commands/todo_registration.py b/loopx/cli_commands/todo_registration.py index 77b7d3c2c9..34023b9984 100644 --- a/loopx/cli_commands/todo_registration.py +++ b/loopx/cli_commands/todo_registration.py @@ -41,6 +41,7 @@ def register_todo_command( "supersede", "archive-completed", "suggest", + "plan", "capture-followups", "project-markdown", ], @@ -50,6 +51,7 @@ def register_todo_command( "agent id, list to read projected todos, update/complete/supersede to transition by todo_id, or " "archive-completed to move older completed todos into Completed Work Archive. " "Use suggest to generate an agent-facing candidate todo analysis prompt without writing state. " + "Use plan with --text and --agent-id for the existing Goal's model planning checkpoint; the caller owns subsequent execution. " "Use capture-followups to record a capped public-safe unclaimed follow-up batch." ), ) diff --git a/loopx/control_plane/goals/start_contract.py b/loopx/control_plane/goals/start_contract.py index ce3579f427..1a395a6323 100644 --- a/loopx/control_plane/goals/start_contract.py +++ b/loopx/control_plane/goals/start_contract.py @@ -5,6 +5,50 @@ GOAL_START_SCHEMA_VERSION = "loopx_goal_start_command_v0" +def goal_planner_contract(*, fine_grained: bool = False) -> dict[str, Any]: + planner = { + "required_before_todo_write": True, + "default_profile": "open_ended_product_direction", + "profile_selection": ( + "Use open_ended_product_direction when the user's goal is a broad, " + "fuzzy product direction or new initiative. Use clear_bounded_problem " + "when the target is a concrete task with a clear success condition. " + "In both cases, let the model produce a real ordered plan before writes." + ), + "profiles": { + "open_ended_product_direction": { + "suggested_items_min": 2, + "suggested_items_max": 5, + "intent": ( + "turn an ambiguous product direction into public-safe, ranked " + "todo options before execution" + ), + }, + "clear_bounded_problem": { + "item_count_policy": "planner_sized", + "may_reuse_current_todo_when_it_already_represents_the_plan": True, + "intent": ( + "make the approach explicit with enough concise ordered todos, " + "without arbitrary caps or management-only filler" + ), + }, + }, + "allowed_priorities": ["P0", "P1", "P2"], + "default_role": "agent", + "default_task_class": "advancement_task", + "required_fields": ["priority", "text", "task_class", "action_kind"], + "public_safe_only": True, + "budget_policy": "minimum sufficient plan; no fixed-count filler", + } + if fine_grained: + planner["fine_grained_plan_horizon"] = ( + "write one current runnable checkpoint; keep later options as evidence-linked " + "planning notes until the existing replan path qualifies the successor" + ) + planner["maximum_runnable_todos_written_ahead"] = 1 + return planner + + def build_goal_start_contract( *, goal_text: str | None, @@ -26,40 +70,7 @@ def build_goal_start_contract( "explicit_invocation_confirms_project_local_state_writes": True, "connect_if_needed": True, "bootstrap_policy": "create project-local LoopX state only when no matching registry goal exists", - "planner": { - "required_before_todo_write": True, - "default_profile": "open_ended_product_direction", - "profile_selection": ( - "Use open_ended_product_direction when the user's goal is a broad, " - "fuzzy product direction or new initiative. Use clear_bounded_problem " - "when the target is a concrete task with a clear success condition. " - "In both cases, let the model produce a real ordered plan before writes." - ), - "profiles": { - "open_ended_product_direction": { - "suggested_items_min": 2, - "suggested_items_max": 5, - "intent": ( - "turn an ambiguous product direction into public-safe, ranked " - "todo options before execution" - ), - }, - "clear_bounded_problem": { - "item_count_policy": "planner_sized", - "may_reuse_current_todo_when_it_already_represents_the_plan": True, - "intent": ( - "make the approach explicit with enough concise ordered todos, " - "without arbitrary caps or management-only filler" - ), - }, - }, - "allowed_priorities": ["P0", "P1", "P2"], - "default_role": "agent", - "default_task_class": "advancement_task", - "required_fields": ["priority", "text", "task_class", "action_kind"], - "public_safe_only": True, - "budget_policy": "minimum sufficient plan; no fixed-count filler", - }, + "planner": goal_planner_contract(fine_grained=fine_grained), "priority_ordering": { "bucket_order": ["P0", "P1", "P2"], "same_priority_tie_breaker": "planner_order_then_todo_write_order", @@ -149,12 +160,6 @@ def build_goal_start_contract( "replan": "direction_change_or_bounded_chain", "checkpoint_accounting": "advancement_only", } - planner = contract["planner"] - planner["fine_grained_plan_horizon"] = ( - "write one current runnable checkpoint; keep later options as evidence-linked " - "planning notes until the existing replan path qualifies the successor" - ) - planner["maximum_runnable_todos_written_ahead"] = 1 return contract diff --git a/loopx/control_plane/goals/start_goal_todo_delta.py b/loopx/control_plane/goals/start_goal_todo_delta.py index 8152323c03..01ff4ea770 100644 --- a/loopx/control_plane/goals/start_goal_todo_delta.py +++ b/loopx/control_plane/goals/start_goal_todo_delta.py @@ -110,10 +110,12 @@ def _todo_add_command_template( runtime_root: str | Path | None, goal_id: str, agent_id: str | None, + registry_path: Path | None = None, ) -> str: return ( f"{render_cli_command_prefix(cli_bin=cli_bin, runtime_root=runtime_root)} " - f"todo add --goal-id " + + (f"--registry {shell_arg(str(registry_path))} " if registry_path is not None else "") + + "todo add --goal-id " f"{shell_arg(str(goal_id or ''))} " "--project . " "--role agent " @@ -136,6 +138,7 @@ def todo_authoring_steps( runtime_root: str | Path | None, goal_id: str, agent_id: str | None, + registry_path: Path | None = None, ) -> list[dict[str, Any]]: """Ordered Todo-authoring steps, conditional on the runnable frontier.""" add_template = _todo_add_command_template( @@ -143,6 +146,7 @@ def todo_authoring_steps( runtime_root=runtime_root, goal_id=goal_id, agent_id=agent_id, + registry_path=registry_path, ) if not existing_runnable_frontier: return [ diff --git a/loopx/control_plane/goals/task_planning.py b/loopx/control_plane/goals/task_planning.py new file mode 100644 index 0000000000..fc76d68255 --- /dev/null +++ b/loopx/control_plane/goals/task_planning.py @@ -0,0 +1,149 @@ +"""Model-owned task planning for an existing Goal, before its caller starts work.""" + +from __future__ import annotations + +import hashlib +import json +import shlex +from pathlib import Path +from typing import Any + +from ...agent_registry import require_registered_agent_id +from ...execution_profile import execution_profile_is_fine_grained +from ...history import load_registry +from ...registry import find_registry_goal +from ...todos import list_goal_todos +from ..todos.todo_semantics import todo_item_is_actionable_open +from ..todos.contract import ( + TODO_STATUS_BLOCKED, + TODO_TASK_CLASS_BLOCKER, + TODO_TASK_CLASS_USER_GATE, +) +from .start_contract import goal_planner_contract +from .start_goal_todo_delta import todo_authoring_steps + + +TASK_PLAN_SCHEMA = "loopx_task_planning_v0" +TASK_PLAN_RESULT_SCHEMA = { + "type": "object", + "additionalProperties": False, + "required": ["input_digest", "status", "todo_ids"], + "properties": { + "input_digest": {"type": "string"}, + "status": {"type": "string", "enum": ["ready", "blocked"]}, + "todo_ids": {"type": "array", "items": {"type": "string"}, "minItems": 1}, + }, +} + + +def build_task_planning_packet( + *, + registry_path: Path, + goal_id: str, + agent_id: str, + text: str, + project: Path | None = None, + runtime_root_arg: str | None = None, +) -> dict[str, Any]: + """Read canonical planning inputs; create no Goal, Todo, Turn or host loop.""" + if not text.strip(): + raise ValueError("todo plan requires non-empty --text") + agent_id = require_registered_agent_id( + registry_path=registry_path, + goal_id=goal_id, + agent_id=agent_id, + field="agent_id", + ) + goal = find_registry_goal(load_registry(registry_path), goal_id) + if goal is None: + raise ValueError("todo plan requires an existing Goal") + # Read both roles explicitly: the compact lane display is not a complete frontier. + todos = [] + for role in ("agent", "user"): + listed = list_goal_todos( + registry_path=registry_path, + goal_id=goal_id, + agent_id=agent_id, + role=role, + project=project, + runtime_root_arg=runtime_root_arg, + ) + todos.extend(listed["todos"]) + runnable = [ + t + for t in todos + if t.get("role") == "agent" + and t.get("task_class") == "advancement_task" + and todo_item_is_actionable_open(t) + ] + fine = execution_profile_is_fine_grained(goal.get("execution_profile")) + prefix = ["loopx", "--format", "json", "--registry", str(registry_path)] + if runtime_root_arg: + prefix += ["--runtime-root", runtime_root_arg] + cli = shlex.join(prefix) + steps = todo_authoring_steps( + existing_runnable_frontier=runnable, + plan_prompt=None, + fine_grained=fine, + cli_bin="loopx", + runtime_root=runtime_root_arg, + goal_id=goal_id, + agent_id=agent_id, + registry_path=registry_path, + ) + identity = {"goal_id": goal_id, "agent_id": agent_id, "text": text} + digest = hashlib.sha256(json.dumps(identity, sort_keys=True).encode()).hexdigest() + return { + "ok": True, + "read_only": True, + "dry_run": True, + "command": "plan", + "schema_version": TASK_PLAN_SCHEMA, + **identity, + "input_digest": digest, + "planner": goal_planner_contract(fine_grained=fine), + "ordered_steps": steps, + "existing_todos": todos, + "runnable_todo_ids": [ + t["todo_id"] for t in runnable if t.get("claimed_by") == agent_id + ], + "blocking_todo_ids": [ + t["todo_id"] + for t in todos + if t.get("status") != "done" + and ( + t.get("status") == TODO_STATUS_BLOCKED + or t.get("task_class") + in {TODO_TASK_CLASS_BLOCKER, TODO_TASK_CLASS_USER_GATE} + ) + ], + "goal_waiting_on": goal.get("waiting_on"), + "result_schema": TASK_PLAN_RESULT_SCHEMA, + "execution_handoff": { + "owner": "caller", + "requires_quota_guard": True, + "starts_host_loop": False, + "spends_quota": False, + "planning_is_advancement": False, + }, + "task_body": ( + "Execute the LoopX task-planning checkpoint for this already registered Goal/Agent. " + "Use the attached planner and ordered_steps, shared with /loopx. Read the exact text " + "and inspect the workspace as needed; make the approach and acceptance explicit " + "before writing task Todos through the routed public CLI. Compare all existing " + "work and waits; reuse/update covered work and add only uncovered work. " + "Do not create a planning/setup Todo, restart the Goal, complete task Todos, " + "change task files, clear waits, activate a loop, execute task work or spend quota. " + "The caller owns the execution_handoff and must enter its quota guard after readback. " + "This is a planning-stage boundary, not task completion. Planning does not grant " + "additional permissions. For ready, return the actual open advancement Todo ids " + "claimed by this agent that cover this input. For blocked, persist/reference the " + "relevant blocker or User gate and return its Todo ids. Return input_digest exactly. " + "Do not claim success from prose or fabricate Todo ids. Command prefix: " + + cli + ), + } + + +def render_task_planning_packet(payload: dict[str, Any]) -> str: + return json.dumps(payload, ensure_ascii=False, indent=2) diff --git a/loopx/slash_command_install.py b/loopx/slash_command_install.py index c9dca36605..c36b124d43 100644 --- a/loopx/slash_command_install.py +++ b/loopx/slash_command_install.py @@ -177,6 +177,7 @@ def _command_prompt_specs(*, cli_bin: str, include_legacy_aliases: bool) -> list host_surface=None, ), "Treat the returned `ordered_steps` and `goal_start_contract` as authoritative. Follow their identity, capability-route, Todo, writeback, host-loop, quota, and stop/gate rules before substantive work; do not reconstruct those rules from skill memory.", + "When the host explicitly supplies a `loopx_task_planning_v0` packet from `loopx todo plan` for a registered Goal/Agent, execute that bounded planning checkpoint instead of starting another Goal. Follow its shared planner and Todo delta, then return actual Todo ids for readback. Its caller-owned execution_handoff retains host activation and quota; do not create a planning Todo, execute task work, or claim delivery during the checkpoint.", "For a Codex App heartbeat, run the returned activation command, require ok=true, and save its `LoopX managed heartbeat bootstrap v2` task_body through automation_update. The saved loader fetches the current thin contract on every wake; do not persist a raw thin/compact/full execution body. Preserve the current goal, registered agent, task binding and existing schedule; read back the automation through the same App.", "If the packet exposes a goal-selection gate, rerun one exact choice before any mutation.", "When authoring task Todos, treat `--action-kind` as the documented extensible public-safe token: choose a short task-relevant value such as `implement`, `test`, or `review`; do not search the LoopX source for an allowlist.", diff --git a/tests/control_plane/test_task_planning.py b/tests/control_plane/test_task_planning.py new file mode 100644 index 0000000000..2192c1e15e --- /dev/null +++ b/tests/control_plane/test_task_planning.py @@ -0,0 +1,123 @@ +from __future__ import annotations + +import contextlib +import io +import json +from pathlib import Path + +import pytest + +from loopx.cli import main +from loopx.control_plane.goals.task_planning import build_task_planning_packet + + +@pytest.fixture +def bound_goal(tmp_path): + state = tmp_path / "state.md" + state.write_text("# Active Goal State\n\n## Agent Todos\n\n## User Todos\n") + registry = tmp_path / "registry.json" + registry.write_text( + json.dumps( + { + "schema_version": "0.1", + "goals": [ + { + "id": "planning-goal", + "repo": str(tmp_path), + "state_file": str(state), + "status": "active", + "waiting_on": "protected approval", + "coordination": {"registered_agents": ["planner", "peer"]}, + } + ], + } + ) + ) + return dict( + registry_path=registry, + goal_id="planning-goal", + agent_id="planner", + text="Investigate the failure, then repair and validate it.\nKeep the API stable.", + project=tmp_path, + runtime_root_arg=str(tmp_path / "runtime"), + ) + + +def test_empty_frontier_plans_before_writes_without_starting_a_loop(bound_goal): + registry = bound_goal["registry_path"] + state = bound_goal["project"] / "state.md" + before = registry.read_bytes(), state.read_bytes() + packet = build_task_planning_packet(**bound_goal) + assert packet["text"] == bound_goal["text"] + assert [step["id"] for step in packet["ordered_steps"]] == [ + "plan_ranked_todos", + "write_ordered_todos", + ] + assert packet["planner"]["required_before_todo_write"] is True + assert packet["execution_handoff"] == { + "owner": "caller", + "requires_quota_guard": True, + "starts_host_loop": False, + "spends_quota": False, + "planning_is_advancement": False, + } + assert packet["goal_waiting_on"] == "protected approval" + assert (registry.read_bytes(), state.read_bytes()) == before + + +def test_existing_plan_is_an_incremental_frontier_not_a_new_goal(bound_goal): + state = bound_goal["project"] / "state.md" + state.write_text("""# Active Goal State + +## Agent Todos +- [ ] [P0] Reproduce the API failure and retain a regression. + +- [ ] [P0] Another agent's task. + + +## User Todos +""") + packet = build_task_planning_packet(**bound_goal) + assert packet["runnable_todo_ids"] == ["todo_existing"] + assert [step["id"] for step in packet["ordered_steps"]] == [ + "compare_planned_todos_with_frontier", + "apply_todo_delta", + ] + assert all(item["todo_id"] != "todo_peer" for item in packet["existing_todos"]) + + +def test_unknown_identity_rejected_without_creating_it(bound_goal): + with pytest.raises(ValueError, match="not registered"): + build_task_planning_packet(**(bound_goal | {"agent_id": "unknown"})) + + +def test_public_cli_returns_a_read_only_checkpoint_and_rejects_execution(bound_goal): + command = [ + "--format", + "json", + "--registry", + str(bound_goal["registry_path"]), + "--runtime-root", + bound_goal["runtime_root_arg"], + "todo", + "plan", + "--goal-id", + bound_goal["goal_id"], + "--agent-id", + bound_goal["agent_id"], + "--project", + str(bound_goal["project"]), + "--text", + bound_goal["text"], + ] + output = io.StringIO() + with contextlib.redirect_stdout(output): + assert main(command) == 0 + packet = json.loads(output.getvalue()) + assert packet["read_only"] and packet["dry_run"] + assert packet["runnable_todo_ids"] == [] + output = io.StringIO() + with contextlib.redirect_stdout(output): + assert main(command + ["--execute"]) == 1 + assert "unsupported" in json.loads(output.getvalue())["error"] + assert not list(Path(bound_goal["runtime_root_arg"]).rglob("*rollout*")) From ee91c6ebdc008e53d1f9d4eff9e9f9193494efa7 Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Fri, 18 Sep 2026 02:39:12 +0800 Subject: [PATCH 2/7] feat(benchmark): add planned and seeded task entry policies Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- benchmark/LHTB/run.sh | 6 +- benchmark/LHTB/scripts/preflight.py | 1 + benchmark/LHTB/scripts/render_config.py | 7 +- benchmark/runtime/codex.py | 6 + benchmark/runtime/harbor.py | 96 ++++-- benchmark/runtime/planning.py | 78 +++++ benchmark/runtime/worker.py | 47 ++- .../configs/shared-heartbeat.yaml | 1 + benchmark/tests/test_shared_codex_runtime.py | 7 +- benchmark/tests/test_task_entry.py | 283 ++++++++++++++++++ 10 files changed, 497 insertions(+), 35 deletions(-) create mode 100644 benchmark/runtime/planning.py create mode 100644 benchmark/tests/test_task_entry.py diff --git a/benchmark/LHTB/run.sh b/benchmark/LHTB/run.sh index 0cc5c5c126..a1ca51be6b 100755 --- a/benchmark/LHTB/run.sh +++ b/benchmark/LHTB/run.sh @@ -47,6 +47,8 @@ LHTB_MODELONLY_GATEWAY="${LHTB_MODELONLY_GATEWAY:-192.0.2.1}" LOOPX_SRC_DIR="${LOOPX_SRC_DIR:-$LOOPX_ROOT}" export PYTHONPATH="$LOOPX_SRC_DIR${PYTHONPATH:+:$PYTHONPATH}" LOOPX_EXECUTION_MODE="${LOOPX_EXECUTION_MODE:-heartbeat}" +LOOPX_TASK_ENTRY="${LOOPX_TASK_ENTRY:-seeded-todo}" +LOOPX_PLANNING_TIMEOUT_SEC="${LOOPX_PLANNING_TIMEOUT_SEC:-300}" LOOPX_ITERATION_CONTEXT="${LOOPX_ITERATION_CONTEXT:-fresh}" LOOPX_VALIDATION_COMMAND_JSON="${LOOPX_VALIDATION_COMMAND_JSON:-[]}" SHARED_CODEX_AGENT_DIR="$LOOPX_SRC_DIR/benchmark/runtime" @@ -115,7 +117,7 @@ if [[ "$MODE" == smoke ]]; then expected_task_count=1 job_suffix="smoke-${SMOKE_TASK}" fi -job_name="lhtb-${LOOPX_EXECUTION_MODE}-${LOOPX_ITERATION_CONTEXT}-${job_suffix}-${run_stamp}" +job_name="lhtb-${LOOPX_EXECUTION_MODE}-${LOOPX_TASK_ENTRY}-${LOOPX_ITERATION_CONTEXT}-${job_suffix}-${run_stamp}" generated_config="$CODE_DIR/.generated/${job_name}.yaml" jobs_dir="$CODE_DIR/runs" @@ -129,6 +131,8 @@ jobs_dir="$CODE_DIR/runs" --effort "$REASONING_EFFORT" \ --timeout "$AGENT_TIMEOUT_SEC" \ --execution-mode "$LOOPX_EXECUTION_MODE" \ + --task-entry "$LOOPX_TASK_ENTRY" \ + --planning-timeout "$LOOPX_PLANNING_TIMEOUT_SEC" \ --iteration-context "$LOOPX_ITERATION_CONTEXT" \ --validation-command-json "$LOOPX_VALIDATION_COMMAND_JSON" \ --turn-timeout "$LOOPX_CODEX_TURN_TIMEOUT_SEC" \ diff --git a/benchmark/LHTB/scripts/preflight.py b/benchmark/LHTB/scripts/preflight.py index a4880f5baf..cc3fd909d7 100755 --- a/benchmark/LHTB/scripts/preflight.py +++ b/benchmark/LHTB/scripts/preflight.py @@ -128,6 +128,7 @@ def check(label: str, passed: bool, detail: str) -> None: ) execution = Execution( mode=kwargs.get("execution_mode", "heartbeat"), + task_entry=kwargs.get("task_entry", "seeded-todo"), context=kwargs.get("iteration_context", "fresh"), validation_command=kwargs.get("validation_command", []), ) diff --git a/benchmark/LHTB/scripts/render_config.py b/benchmark/LHTB/scripts/render_config.py index b9d02ed3f2..b27023536e 100755 --- a/benchmark/LHTB/scripts/render_config.py +++ b/benchmark/LHTB/scripts/render_config.py @@ -9,7 +9,7 @@ from pathlib import Path import yaml -from benchmark.runtime.codex import CONTEXTS, MODES, Execution +from benchmark.runtime.codex import CONTEXTS, MODES, TASK_ENTRIES, Execution def main() -> int: @@ -25,6 +25,8 @@ def main() -> int: parser.add_argument("--task", action="append", default=[]) parser.add_argument("--execution-mode", choices=MODES, default="heartbeat") parser.add_argument("--iteration-context", choices=CONTEXTS, default="fresh") + parser.add_argument("--task-entry", choices=TASK_ENTRIES, default="seeded-todo") + parser.add_argument("--planning-timeout", type=float, default=300) parser.add_argument("--validation-command-json", default="[]") parser.add_argument("--turn-timeout", type=float, default=4700) parser.add_argument("--scheduler-timeout", type=int, default=5080) @@ -59,6 +61,7 @@ def main() -> int: context=args.iteration_context, timeout_seconds=args.turn_timeout, validation_command=json.loads(args.validation_command_json), + task_entry=args.task_entry, ) agent["kwargs"].update( execution_mode=execution.mode, @@ -66,6 +69,8 @@ def main() -> int: validation_command=list(execution.validation_command), turn_timeout_sec=execution.timeout_seconds, scheduler_timeout_sec=args.scheduler_timeout, + task_entry=execution.task_entry, + planning_timeout_sec=args.planning_timeout, ) agent["kwargs"]["goals"] = str(execution.native_goal).lower() agent["kwargs"]["web_search"] = "disabled" diff --git a/benchmark/runtime/codex.py b/benchmark/runtime/codex.py index e41d7b4dc3..4da2ff4fe7 100644 --- a/benchmark/runtime/codex.py +++ b/benchmark/runtime/codex.py @@ -11,6 +11,7 @@ MODES = ("plain", "native-goal", "heartbeat", "turn", "loopx-goal") CONTEXTS = ("fresh", "resume-if-available") +TASK_ENTRIES = ("seeded-todo", "loopx-planned") SANDBOXES = ("read-only", "workspace-write", "danger-full-access") @@ -21,10 +22,15 @@ class Execution: sandbox: str = "danger-full-access" timeout_seconds: float = 4700 validation_command: tuple[str, ...] = () + task_entry: str = "seeded-todo" def __post_init__(self) -> None: if self.mode not in MODES or self.context not in CONTEXTS: raise ValueError("unsupported execution mode or iteration context") + if self.task_entry not in TASK_ENTRIES: + raise ValueError("unsupported task entry") + if self.task_entry == "loopx-planned" and not self.uses_loopx: + raise ValueError("loopx-planned requires a LoopX execution mode") if self.context != "fresh" and self.mode != "turn": raise ValueError("resume-if-available currently requires mode=turn") if self.sandbox not in SANDBOXES: diff --git a/benchmark/runtime/harbor.py b/benchmark/runtime/harbor.py index fd60c54081..f50ac779a3 100644 --- a/benchmark/runtime/harbor.py +++ b/benchmark/runtime/harbor.py @@ -7,6 +7,7 @@ import shlex import subprocess import tempfile +import time from pathlib import Path from typing import Iterable @@ -55,6 +56,8 @@ def __init__( turn_timeout_sec=4700, scheduler_timeout_sec=5080, replan_after_todos=3, + task_entry="seeded-todo", + planning_timeout_sec=300, **kwargs, ): if isinstance(validation_command, str): @@ -65,7 +68,11 @@ def __init__( codex_sandbox, float(turn_timeout_sec), validation_command if validation_command is not None else (), + task_entry, ) + self.planning_timeout = float(planning_timeout_sec) + if not 0 < self.planning_timeout < float("inf"): + raise ValueError("planning timeout must be finite and positive") self.scheduler_timeout = int(scheduler_timeout_sec) if self.scheduler_timeout <= self.execution.timeout_seconds + 150: raise ValueError( @@ -215,6 +222,7 @@ async def install(self, environment: BaseEnvironment) -> None: "runtime_profile": "generic_cli", "execution_mode": self.execution.mode, "iteration_context": self.execution.context, + "task_entry": self.execution.task_entry, "home_scope": "trial", "login_shell_node_path": _BASH_ENV, "scheduler_terminal_packet_compatibility": True, @@ -239,13 +247,19 @@ async def _write_task_document( handle.write(instruction.strip()) handle.write("\n") await environment.upload_file(Path(name), _TASK_DOC) + await environment.upload_file(Path(name), self._task_document) await self.exec_as_root( environment, - command=f"chmod 0644 {_TASK_DOC}", + command=f"chmod 0644 {_TASK_DOC} {self._task_document}", ) finally: Path(name).unlink(missing_ok=True) + @property + def _task_document(self) -> str: + # Existing Todos keep their original input when Harbor supplies a new phase. + return f"{_CONTROL}/task-phase-{self._phase_number:03d}.md" + async def _loopx( self, environment: BaseEnvironment, @@ -291,6 +305,11 @@ async def _registry_exists(self, environment: BaseEnvironment) -> bool: async def _prepare_phase( self, environment: BaseEnvironment, instruction: str, *, cwd: str ) -> None: + pending = await environment.exec( + command=f"test -e {_LOOPX_RUNTIME}/benchmark-pending-turn.json" + ) + if pending.return_code == 0: + raise RuntimeError("Resolve the pending Turn before entering another task phase") await self._write_task_document(environment, instruction) if not await self._registry_exists(environment): await self._loopx( @@ -338,7 +357,7 @@ async def _prepare_phase( cwd=cwd, ) else: - # Harbor invoked a new task phase; this is not an automatic unblock. + # New input is not evidence that an existing wait or gate was resolved. await self._loopx( environment, [ @@ -347,14 +366,25 @@ async def _prepare_phase( _GOAL_ID, "--execution-replan-after-todos", str(self.replan_after_todos), - "--clear-waiting-on", - "--agent-work-mode", - f"{_AGENT_ID}=active", "--execute", ], cwd=cwd, ) + if self.execution.task_entry == "seeded-todo": + await self._seed_phase(environment, cwd=cwd) + + cadence = await self._loopx( + environment, ["configure-goal", "--goal-id", _GOAL_ID], cwd=cwd, + ) + configured_state = cadence.get("after") or cadence.get("before") or {} + configured = configured_state.get("execution_profile", {}).get("replan_after_completed_todos") + if configured != self.replan_after_todos: + raise RuntimeError( + f"replan cadence readback mismatch: expected {self.replan_after_todos}, got {configured!r}" + ) + + async def _seed_phase(self, environment: BaseEnvironment, *, cwd: str) -> None: todo_id = f"benchmark-task-phase-{self._phase_number:03d}" await self._loopx( environment, @@ -370,7 +400,7 @@ async def _prepare_phase( "--text", ( f"[P0] Execute benchmark phase {self._phase_number}. Read the exact " - f"current task from {_TASK_DOC}; inspect the workspace, implement and " + f"current task from {self._task_document}; inspect the workspace, implement and " "validate it, and create bounded successor Todos for remaining work." ), "--task-class", @@ -386,20 +416,6 @@ async def _prepare_phase( cwd=cwd, ) - cadence = await self._loopx( - environment, - ["configure-goal", "--goal-id", _GOAL_ID], - cwd=cwd, - ) - configured_state = cadence.get("after") or cadence.get("before") or {} - configured = configured_state.get("execution_profile", {}).get( - "replan_after_completed_todos" - ) - if configured != self.replan_after_todos: - raise RuntimeError( - f"replan cadence readback mismatch: expected {self.replan_after_todos}, got {configured!r}" - ) - def _worker_env(self, *, cwd: str) -> dict[str, str]: env = self._profile_env() if not self.execution.uses_loopx: @@ -413,11 +429,12 @@ def _worker_env(self, *, cwd: str) -> dict[str, str]: "LOOPX_GOAL_ID": _GOAL_ID, "LOOPX_AGENT_ID": _AGENT_ID, "LOOPX_PROJECT": cwd, - "LOOPX_TASK_DOC": _TASK_DOC, + "LOOPX_TASK_DOC": self._task_document, "LOOPX_WAKE_LOG_DIR": _WAKE_LOG_DIR, "LOOPX_CODEX_HOME": _CODEX_HOME, "LOOPX_SHARED_SKILLS": _SHARED_SKILLS, "LOOPX_EXECUTION_MODE": self.execution.mode, + "LOOPX_TASK_ENTRY": self.execution.task_entry, "LOOPX_ITERATION_CONTEXT": self.execution.context, "LOOPX_CODEX_SANDBOX": self.execution.sandbox, "LOOPX_VALIDATION_COMMAND_JSON": json.dumps( @@ -524,6 +541,7 @@ def _populate_context( context.metadata = { "execution_mode": self.execution.mode, "iteration_context": self.execution.context, + "task_entry": self.execution.task_entry, "home_scope": "trial", "replan_after_completed_todos": self.replan_after_todos, "benchmark_phase": self._phase_number, @@ -543,6 +561,7 @@ async def run( if not self.model_name: raise ValueError("model_name is required") self._phase_number += 1 + deadline = time.monotonic() + self.scheduler_timeout pwd = await self.exec_as_agent(environment, command="pwd", timeout_sec=30) cwd = (pwd.stdout or "").strip() if not cwd.startswith("/"): @@ -557,6 +576,33 @@ async def run( else: await self._write_task_document(environment, instruction) wake_command = [f"{_PYTHON}/bin/python3", "-m", _WORKER_MODULE] + env = self._worker_env(cwd=cwd) + if self.execution.task_entry == "loopx-planned": + result_path = f"{_CONTROL}/planning-phase-{self._phase_number:03d}.json" + planning_timeout = min(self.planning_timeout, deadline - time.monotonic() - 30) + if planning_timeout <= 0: + raise TimeoutError("Task budget exhausted before planning") + await self.exec_as_agent( + environment, command=shlex.join(wake_command), cwd=cwd, + env=env | { + "LOOPX_TASK_STAGE": "plan", + "LOOPX_PLANNING_TIMEOUT_SEC": str(planning_timeout), + "LOOPX_PLANNING_RESULT": result_path, + }, + timeout_sec=planning_timeout + 30, + ) + observed = await environment.exec(command=f"cat {result_path}") + entry = json.loads(observed.stdout or "") + if entry.get("state_readback_verified") is not True: + raise RuntimeError("Planning did not return verified Todo readback") + if entry["status"] == "blocked": + return + remaining = int(deadline - time.monotonic()) + if remaining <= 150: + raise TimeoutError("Task budget exhausted before execution handoff") + # Planning consumes the phase budget, including when the host later resumes. + host_timeout = min(self.execution.timeout_seconds, remaining - 150) + env["LOOPX_CODEX_TURN_TIMEOUT_SEC"] = str(host_timeout) if self.execution.mode in {"heartbeat", "turn"}: command = [ f"{_PYTHON}/bin/python3", @@ -578,7 +624,7 @@ async def run( "--wake-cmd", "exec " + shlex.join(wake_command), "--wake-timeout-seconds", - str(self.execution.timeout_seconds + 150), + str(host_timeout + 150), "--quota-timeout-seconds", "30", "--error-backoff-seconds", @@ -589,7 +635,7 @@ async def run( phase_log = f"/logs/agent/worker-phase-{self._phase_number:03d}.log" shell = ( "set +e; " - f"timeout --signal=TERM --kill-after=30 {self.scheduler_timeout}s " + f"timeout --signal=TERM --kill-after=30 {remaining}s " f"{shlex.join(command)} >> {shlex.quote(phase_log)} 2>&1; " "rc=$?; " # Budget exhaustion retains partial artifacts for native scoring. @@ -598,9 +644,9 @@ async def run( await self.exec_as_agent( environment, command=shell, - env=self._worker_env(cwd=cwd), + env=env, cwd=cwd, - timeout_sec=self.scheduler_timeout + 60, + timeout_sec=remaining + 60, ) finally: # Remote Harbor backends download logs after run(). Read them now diff --git a/benchmark/runtime/planning.py b/benchmark/runtime/planning.py new file mode 100644 index 0000000000..0d558c8e6b --- /dev/null +++ b/benchmark/runtime/planning.py @@ -0,0 +1,78 @@ +"""Consume the product task-planning checkpoint and verify its state readback.""" + +from __future__ import annotations + +import json +import subprocess +from pathlib import Path + + +def task_plan_packet(env: dict[str, str], cli: list[str]) -> dict: + command = cli + [ + "todo", + "plan", + "--goal-id", + env["LOOPX_GOAL_ID"], + "--agent-id", + env["LOOPX_AGENT_ID"], + "--project", + env["LOOPX_PROJECT"], + "--text", + Path(env["LOOPX_TASK_DOC"]).read_text(encoding="utf-8"), + ] + response = subprocess.run( + command, + cwd=env["LOOPX_PROJECT"], + env=env, + text=True, + capture_output=True, + check=True, + timeout=120, + ) + packet = json.loads(response.stdout) + if ( + packet.get("ok") is not True + or packet.get("schema_version") != "loopx_task_planning_v0" + or packet.get("goal_id") != env["LOOPX_GOAL_ID"] + or packet.get("agent_id") != env["LOOPX_AGENT_ID"] + or packet.get("execution_handoff", {}).get("owner") != "caller" + ): + raise ValueError("product planning packet does not match the caller binding") + return packet + + +def validate_plan_readback(result: dict, before: dict, after: dict) -> dict: + if ( + result.get("input_digest") != before["input_digest"] + or after["input_digest"] != before["input_digest"] + ): + raise ValueError("planning input changed before readback") + status = result.get("status") + ids = result.get("todo_ids") + if ( + status not in {"ready", "blocked"} + or not isinstance(ids, list) + or not ids + or any(not isinstance(item, str) for item in ids) + or len(set(ids)) != len(ids) + ): + raise ValueError( + "planning result requires unique actual Todo ids and a typed status" + ) + todos = {item["todo_id"]: item for item in after["existing_todos"]} + if any(todo_id not in todos for todo_id in ids): + raise ValueError("planning referenced a missing or unrelated Todo") + if status == "ready" and not set(ids).issubset(after["runnable_todo_ids"]): + raise ValueError("planning referenced non-runnable or unclaimed work") + if status == "blocked" and not set(ids).issubset(after["blocking_todo_ids"]): + raise ValueError("blocked planning result requires unresolved blocking Todos") + return { + "input_digest": before["input_digest"], + "status": status, + "todo_ids": ids, + "goal_id": after["goal_id"], + "agent_id": after["agent_id"], + "state_readback_verified": True, + "execution_owner": "caller", + "planning_session_reused_for_execution": False, + } diff --git a/benchmark/runtime/worker.py b/benchmark/runtime/worker.py index fdaee65e5d..f9f969f253 100644 --- a/benchmark/runtime/worker.py +++ b/benchmark/runtime/worker.py @@ -206,7 +206,11 @@ def run_once(env: dict[str, str]) -> dict: sandbox=env.get("LOOPX_CODEX_SANDBOX", "danger-full-access"), timeout_seconds=float(env.get("LOOPX_CODEX_TURN_TIMEOUT_SEC", "4700")), validation_command=json.loads(env.get("LOOPX_VALIDATION_COMMAND_JSON", "[]")), + task_entry=env.get("LOOPX_TASK_ENTRY", "seeded-todo"), ) + stage = env.get("LOOPX_TASK_STAGE", "execute") + if stage not in {"plan", "execute"} or (stage == "plan" and execution.task_entry != "loopx-planned"): + raise ValueError("invalid task-entry stage") turn_id = f"wake-{time.time_ns()}-{uuid.uuid4().hex[:12]}" log_root = Path(env["LOOPX_WAKE_LOG_DIR"]) wake = log_root / turn_id @@ -218,6 +222,8 @@ def run_once(env: dict[str, str]) -> dict: "turn_id": turn_id, "mode": execution.mode, "context": execution.context, + "task_entry": execution.task_entry, + "stage": stage, "home_scope": "trial", "ok": False, "timed_out": False, @@ -238,16 +244,23 @@ def run_once(env: dict[str, str]) -> dict: skills=Path(env["LOOPX_SHARED_SKILLS"]) if execution.uses_loopx else None, ) body = "Finish the task." - if execution.mode in {"heartbeat", "loopx-goal"}: + if stage == "plan": + from benchmark.runtime.planning import task_plan_packet + + planning_before = task_plan_packet(env, loopx_command(env)) + body = "$loopx\n\nHost-supplied planning checkpoint:\n" + json.dumps(planning_before, ensure_ascii=False) + (wake / "planning-input.json").write_text(body, encoding="utf-8") + (wake / "planning-schema.json").write_text(json.dumps(planning_before["result_schema"])) + elif execution.mode in {"heartbeat", "loopx-goal"}: body = heartbeat_body(env, turn_id, native_goal=execution.native_goal) elif execution.mode == "plain": body = Path(env["LOOPX_TASK_DOC"]).read_text(encoding="utf-8") with (wake / "stderr.log").open("w") as stderr: - if execution.native_goal: + if execution.native_goal and stage == "execute": run_native_goal(env, execution, body, receipt, stderr) else: pending = {} - if execution.mode == "turn": + if execution.mode == "turn" and stage == "execute": pending_path.parent.mkdir(parents=True, exist_ok=True) if pending_path.exists(): pending = read_json(pending_path.read_text()) @@ -261,7 +274,7 @@ def run_once(env: dict[str, str]) -> dict: pending["turn_instance_id"], pending.get("resume_turn_key"), ) - if execution.mode == "turn" + if execution.mode == "turn" and stage == "execute" else [ env["CODEX_BIN"], "exec", @@ -271,6 +284,11 @@ def run_once(env: dict[str, str]) -> dict: execution.sandbox, "--cd", env["LOOPX_PROJECT"], + *([ + "-c", "features.goals=false", + "--output-schema", str(wake / "planning-schema.json"), + "--output-last-message", str(wake / "planning-result.json"), + ] if stage == "plan" else []), "-", ] ) @@ -278,13 +296,26 @@ def run_once(env: dict[str, str]) -> dict: with child_process( command, env=env, stdout=stdout, stderr=stderr ) as process: - allowance = 150 if execution.mode == "turn" else 0 + allowance = 150 if execution.mode == "turn" and stage == "execute" else 0 + timeout = (float(env["LOOPX_PLANNING_TIMEOUT_SEC"]) if stage == "plan" + else execution.timeout_seconds + allowance) process.communicate( - input=body, timeout=execution.timeout_seconds + allowance + input=body, timeout=timeout ) receipt["return_code"] = process.returncode receipt["ok"] = process.returncode == 0 - if execution.mode == "turn": + if stage == "plan" and receipt["ok"]: + from benchmark.runtime.planning import task_plan_packet, validate_plan_readback + + result = read_json((wake / "planning-result.json").read_text()) + receipt["planning"] = validate_plan_readback( + result, planning_before, task_plan_packet(env, loopx_command(env)), + ) + target = Path(env["LOOPX_PLANNING_RESULT"]) + temporary = target.with_suffix(".tmp") + temporary.write_text(json.dumps(receipt["planning"])) + temporary.replace(target) + elif execution.mode == "turn" and stage == "execute": result = read_json((wake / "stdout.jsonl").read_text()) receipt["turn_execution"] = result receipt["ok"] = receipt["ok"] and result.get("ok") is True @@ -298,8 +329,10 @@ def run_once(env: dict[str, str]) -> dict: temporary.write_text(json.dumps(pending)) temporary.replace(pending_path) except subprocess.TimeoutExpired: + receipt["ok"] = False receipt["timed_out"] = True except BaseException as exc: + receipt["ok"] = False receipt["error_kind"] = type(exc).__name__ raise finally: diff --git a/benchmark/swe-marathon/configs/shared-heartbeat.yaml b/benchmark/swe-marathon/configs/shared-heartbeat.yaml index 3ef35a52e5..da9300877f 100644 --- a/benchmark/swe-marathon/configs/shared-heartbeat.yaml +++ b/benchmark/swe-marathon/configs/shared-heartbeat.yaml @@ -6,6 +6,7 @@ agents: override_timeout_sec: 5400 kwargs: execution_mode: heartbeat + task_entry: seeded-todo # Use loopx-planned for the product planning checkpoint. iteration_context: fresh reasoning_effort: high turn_timeout_sec: 4700 diff --git a/benchmark/tests/test_shared_codex_runtime.py b/benchmark/tests/test_shared_codex_runtime.py index 659240d7ea..fcbcaa7e5b 100644 --- a/benchmark/tests/test_shared_codex_runtime.py +++ b/benchmark/tests/test_shared_codex_runtime.py @@ -8,6 +8,7 @@ import sys import time import tomllib +from types import SimpleNamespace from pathlib import Path import pytest @@ -361,9 +362,13 @@ async def cli(environment, args, **kwargs): monkeypatch.setattr(agent, "_write_task_document", write_task) monkeypatch.setattr(agent, "_registry_exists", registry_exists) monkeypatch.setattr(agent, "_loopx", cli) - asyncio.run(agent._prepare_phase(None, "Synthetic task", cwd=str(tmp_path))) + async def no_pending(**kwargs): + return SimpleNamespace(return_code=1) + + asyncio.run(agent._prepare_phase(SimpleNamespace(exec=no_pending), "Synthetic task", cwd=str(tmp_path))) assert any(args[:2] == ["todo", "add"] for args in calls) assert any(args[0] == "bootstrap" for args in calls) is not existing + assert all("--clear-waiting-on" not in args for args in calls) def test_staged_snapshot_keeps_observed_commit_when_branch_moves(tmp_path, monkeypatch): diff --git a/benchmark/tests/test_task_entry.py b/benchmark/tests/test_task_entry.py new file mode 100644 index 0000000000..cf2501ca62 --- /dev/null +++ b/benchmark/tests/test_task_entry.py @@ -0,0 +1,283 @@ +from __future__ import annotations + +import asyncio +import json +import os +import sys +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from benchmark.runtime.codex import Execution +from benchmark.runtime.planning import validate_plan_readback +from benchmark.runtime.worker import run_once + + +@pytest.mark.parametrize( + "kwargs", + [ + {"task_entry": "unknown"}, + {"mode": "plain", "task_entry": "loopx-planned"}, + {"mode": "native-goal", "task_entry": "loopx-planned"}, + ], +) +def test_invalid_entry_rejected_before_model_call(kwargs): + with pytest.raises(ValueError): + Execution(**kwargs) + + +@pytest.fixture +def planning_env(tmp_path): + project = tmp_path / "project" + project.mkdir() + task = tmp_path / "task.md" + task.write_text("Repair the failure and preserve a regression test.\n") + state = tmp_path / "state.md" + state.write_text("# Active Goal State\n\n## Agent Todos\n\n## User Todos\n") + registry = tmp_path / "registry.json" + registry.write_text( + json.dumps( + { + "schema_version": "0.1", + "goals": [ + { + "id": "planning-goal", + "repo": str(project), + "state_file": str(state), + "status": "active", + "coordination": {"registered_agents": ["planner"]}, + } + ], + } + ) + ) + cli = tmp_path / "loopx" + cli.write_text( + f"#!{sys.executable}\nfrom loopx.cli import main\nraise SystemExit(main())\n" + ) + cli.chmod(0o755) + skills = tmp_path / "skills" + skills.mkdir() + binary = tmp_path / "codex" + binary.write_text( + f"#!{sys.executable}\n" + + """ +import json, os, pathlib, subprocess, sys +packet = json.loads(sys.stdin.read().split("Host-supplied planning checkpoint:\\n", 1)[1]) +command = [os.environ["LOOPX_CLI"], "--format", "json", "--registry", os.environ["LOOPX_REGISTRY"], + "--runtime-root", os.environ["LOOPX_RUNTIME_ROOT"], "todo", "add", "--goal-id", "planning-goal", + "--role", "agent", "--claimed-by", "planner", + "--text", "[P0] Reproduce and repair the failure, then pass the regression.", + "--task-class", "advancement_task", "--action-kind", "implement", "--execute"] +if not packet["runnable_todo_ids"]: + written = json.loads(subprocess.run(command, check=True, capture_output=True, text=True).stdout) + todo_id = written["todo_id"] +else: + todo_id = packet["runnable_todo_ids"][0] +result = {"input_digest": packet["input_digest"], "status": "ready", "todo_ids": [todo_id]} +pathlib.Path(sys.argv[sys.argv.index("--output-last-message") + 1]).write_text(json.dumps(result)) +pathlib.Path(os.environ["CODEX_HOME"], "seen-argv.json").write_text(json.dumps(sys.argv)) +""" + ) + binary.chmod(0o755) + return dict(os.environ) | { + "PYTHONPATH": str(Path(__file__).resolve().parents[2]), + "LOOPX_EXECUTION_MODE": "heartbeat", + "LOOPX_TASK_ENTRY": "loopx-planned", + "LOOPX_TASK_STAGE": "plan", + "LOOPX_PLANNING_TIMEOUT_SEC": "30", + "LOOPX_PLANNING_RESULT": str(tmp_path / "planning.json"), + "LOOPX_GOAL_ID": "planning-goal", + "LOOPX_AGENT_ID": "planner", + "LOOPX_REGISTRY": str(registry), + "LOOPX_RUNTIME_ROOT": str(tmp_path / "runtime"), + "LOOPX_CLI": str(cli), + "CODEX_BIN": str(binary), + "LOOPX_PROJECT": str(project), + "LOOPX_TASK_DOC": str(task), + "LOOPX_CODEX_HOME": str(tmp_path / "home"), + "LOOPX_SHARED_SKILLS": str(skills), + "LOOPX_WAKE_LOG_DIR": str(tmp_path / "logs" / "wakes"), + "MODEL_NAME": "fixture", + "REASONING_EFFORT": "high", + "OPENAI_BASE_URL": "http://localhost:8123/v1", + "OPENAI_API_KEY": "fixture-key", + } + + +def test_planning_writes_real_todo_then_reuses_it_without_executing_task(planning_env): + ids = [] + for _ in range(2): + receipt = run_once(planning_env) + assert receipt["ok"] and receipt["planning"]["state_readback_verified"] + ids.append(receipt["planning"]["todo_ids"]) + state = Path(planning_env["LOOPX_REGISTRY"]).with_name("state.md").read_text() + assert ids[0] == ids[1] and len(ids[0]) == 1 + assert state.count("todo_id=" + ids[0][0]) == 1 + assert not list(Path(planning_env["LOOPX_PROJECT"]).iterdir()) + argv = json.loads( + (Path(planning_env["LOOPX_CODEX_HOME"]) / "seen-argv.json").read_text() + ) + assert "features.goals=false" in argv and "resume" not in argv + assert not list( + Path(planning_env["LOOPX_RUNTIME_ROOT"]).rglob("benchmark-pending-turn.json") + ) + + +def test_prose_or_fabricated_todo_does_not_qualify_planning(planning_env): + binary = Path(planning_env["CODEX_BIN"]) + binary.write_text( + binary.read_text().replace( + '"todo_ids": [todo_id]', '"todo_ids": ["todo_fabricated"]' + ) + ) + with pytest.raises(ValueError, match="missing or unrelated"): + run_once(planning_env) + assert not Path(planning_env["LOOPX_PLANNING_RESULT"]).exists() + receipt = json.loads( + next( + Path(planning_env["LOOPX_WAKE_LOG_DIR"]).glob("*/receipt.json") + ).read_text() + ) + # A zero host exit must not be mistaken for qualified planning. + assert ( + not receipt["ok"] + and receipt.get("planning") is None + and receipt["error_kind"] == "ValueError" + ) + + +def test_plan_readback_rejects_stale_input_unclaimed_work_and_false_blockers(): + packet = { + "input_digest": "input-a", + "goal_id": "goal", + "agent_id": "agent", + "existing_todos": [ + {"todo_id": "todo_owned"}, + {"todo_id": "todo_unclaimed"}, + {"todo_id": "todo_gate"}, + ], + "runnable_todo_ids": ["todo_owned"], + "blocking_todo_ids": ["todo_gate"], + } + result = {"input_digest": "input-a", "status": "ready", "todo_ids": ["todo_owned"]} + assert validate_plan_readback(result, packet, packet)["status"] == "ready" + assert ( + validate_plan_readback( + result | {"status": "blocked", "todo_ids": ["todo_gate"]}, packet, packet + )["status"] + == "blocked" + ) + for invalid in [ + result | {"input_digest": "old"}, + result | {"todo_ids": ["todo_unclaimed"]}, + result | {"status": "blocked"}, + result | {"todo_ids": ["todo_owned", "todo_owned"]}, + ]: + with pytest.raises(ValueError): + validate_plan_readback(invalid, packet, packet) + + +def test_planned_phase_preserves_waits_and_does_not_prewrite_a_todo( + tmp_path, monkeypatch +): + pytest.importorskip("harbor") + from benchmark.runtime.harbor import BenchmarkCodex + + agent = BenchmarkCodex( + logs_dir=tmp_path, model_name="openai/fixture", task_entry="loopx-planned" + ) + calls = [] + + async def prepared(*args, **kwargs): + return True + + async def cli(environment, args, **kwargs): + calls.append(args) + return {"after": {"execution_profile": {"replan_after_completed_todos": 3}}} + + async def no_pending(**kwargs): + return SimpleNamespace(return_code=1) + + monkeypatch.setattr(agent, "_write_task_document", prepared) + monkeypatch.setattr(agent, "_registry_exists", prepared) + monkeypatch.setattr(agent, "_loopx", cli) + asyncio.run( + agent._prepare_phase( + SimpleNamespace(exec=no_pending), "new feedback", cwd=str(tmp_path) + ) + ) + assert all(args[:2] != ["todo", "add"] for args in calls) + assert all( + "--clear-waiting-on" not in args and "--agent-work-mode" not in args + for args in calls + ) + + +@pytest.mark.parametrize("status", ["ready", "blocked"]) +def test_planning_budget_and_blocked_handoff_use_the_real_adapter_run( + tmp_path, monkeypatch, status +): + pytest.importorskip("harbor") + from benchmark.runtime import harbor + + agent = harbor.BenchmarkCodex( + logs_dir=tmp_path, + model_name="openai/fixture", + task_entry="loopx-planned", + turn_timeout_sec=250, + scheduler_timeout_sec=500, + ) + clock = [0.0] + executions = [] + + async def prepare(*args, **kwargs): + pass + + async def execute(environment, *, command, env=None, **kwargs): + if command == "pwd": + return SimpleNamespace(stdout="/workspace", return_code=0) + if env.get("LOOPX_TASK_STAGE") == "plan": + clock[0] = 200 + else: + executions.append((command, env)) + return SimpleNamespace(stdout="", return_code=0) + + async def read_result(**kwargs): + return SimpleNamespace( + stdout=json.dumps({"status": status, "state_readback_verified": True}) + ) + + monkeypatch.setattr(harbor.time, "monotonic", lambda: clock[0]) + monkeypatch.setattr(agent, "_prepare_phase", prepare) + monkeypatch.setattr(agent, "exec_as_agent", execute) + monkeypatch.setattr(agent, "_populate_context", lambda *args: None) + environment = SimpleNamespace(exec=read_result, is_mounted=True) + asyncio.run(agent.run("Synthetic task", environment, SimpleNamespace())) + assert len(executions) == (1 if status == "ready" else 0) + if executions: + command, env = executions[0] + assert "--kill-after=30 300s" in command + assert float(env["LOOPX_CODEX_TURN_TIMEOUT_SEC"]) == 150 + + +def test_pending_turn_prevents_phase_input_replacement(tmp_path, monkeypatch): + pytest.importorskip("harbor") + from benchmark.runtime.harbor import BenchmarkCodex + + agent = BenchmarkCodex(logs_dir=tmp_path, model_name="openai/fixture") + + async def pending(**kwargs): + return SimpleNamespace(return_code=0) + + async def unexpected_write(*args, **kwargs): + pytest.fail("pending transaction input must not be replaced") + + monkeypatch.setattr(agent, "_write_task_document", unexpected_write) + with pytest.raises(RuntimeError, match="pending Turn"): + asyncio.run( + agent._prepare_phase( + SimpleNamespace(exec=pending), "next task", cwd=str(tmp_path) + ) + ) From 5156663cc9cc4e95ace8d5df1349b9337567fd4e Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Fri, 18 Sep 2026 02:39:12 +0800 Subject: [PATCH 3/7] docs(planning): describe task-entry ablation and driver handoff Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- benchmark/LHTB/README.md | 7 +++ benchmark/runtime/RUNTIME.md | 49 ++++++++++++++++++- ...n-harness-benchmark-research-program-v0.md | 8 +++ ...ess-benchmark-research-program-v0.zh-CN.md | 6 +++ docs/project-agent-todo-contract.md | 23 +++++++++ 5 files changed, 92 insertions(+), 1 deletion(-) diff --git a/benchmark/LHTB/README.md b/benchmark/LHTB/README.md index d91a916a5b..8d8bb5f582 100644 --- a/benchmark/LHTB/README.md +++ b/benchmark/LHTB/README.md @@ -118,6 +118,13 @@ accepts `resume-if-available`. Turn also requires `LOOPX_VALIDATION_COMMAND_JSON an argv array for the independently protected task validator. No generic benchmark scoring or hidden-verifier feedback is introduced. +`LOOPX_TASK_ENTRY=seeded-todo` preserves the generic phase Todo default. +`LOOPX_TASK_ENTRY=loopx-planned` invokes the product planning checkpoint before +heartbeat, Turn or LoopX Goal execution. `LOOPX_PLANNING_TIMEOUT_SEC` defaults +to 300 and consumes the existing phase budget. Both entry policies preserve +existing waits when new phases arrive. See the shared runtime for session and +planning-readback semantics. + Model and effort defaults remain unchanged but may be selected explicitly. `run.sh prepare` performs networking/Harbor preparation. `preflight` now checks existing preparation without patching Harbor or creating a network. Smoke/full diff --git a/benchmark/runtime/RUNTIME.md b/benchmark/runtime/RUNTIME.md index 43eb81c917..689dab9c8e 100644 --- a/benchmark/runtime/RUNTIME.md +++ b/benchmark/runtime/RUNTIME.md @@ -17,6 +17,7 @@ agents: override_timeout_sec: 5400 kwargs: execution_mode: heartbeat + task_entry: seeded-todo iteration_context: fresh reasoning_effort: max codex_sandbox: danger-full-access @@ -46,6 +47,52 @@ The runner supplies no HEAD-moved/clean-worktree/exit-only substitute and never calls hidden benchmark verification to provide intermediate feedback. Independent validator protection remains the environment owner's responsibility. +## Task entry and planning ablation + +`task_entry` is independent of the execution mode: + +- `seeded-todo` (the compatibility default) writes one generic execution Todo + per native phase. The agent can still plan and replan during execution. +- `loopx-planned` runs the installed `$loopx` skill against the public + `loopx todo plan` checkpoint before execution. The checkpoint shares the + product's planner and continuation-aware Todo delta; it creates no planning + Todo and starts no host loop. Select it only for heartbeat, Turn or LoopX Goal. + +The model writes or reuses actual task Todos through the public CLI. The worker +reads the product packet again and checks the input digest, identity, Todo ids +and runnable/blocked state. A fabricated id, changed input, wrong owner, failed +planning process or missing result fails the entry; it never falls back to a +generic Todo. A blocked entry retains the referenced blockers and starts no +execution driver. Readback proves state and ownership, not semantic plan quality. + +Planning uses a separate fresh `codex exec` session with native Goals disabled +for that call. Its session is not inserted into core Turn session bindings or +resumed by the subsequent execution. This is a planning-contract ablation, not +an exact reproduction of same-conversation interactive `$loopx` startup. +The default `planning_timeout_sec` is 300; planning and preparation consume the +same `scheduler_timeout_sec` phase budget as execution. Planning sessions are +included in native session/token aggregation. No planning checkpoint is counted +as a completed advancement Todo or settled work Turn. + +Each phase keeps an immutable task document. New phases preserve Goal/Agent +identity and expose existing Todos to the planner; they do not clear waiting +state or force the agent active. An unresolved Turn must be recovered before +another phase can replace its task input. These wait/recovery rules apply to +both entry policies; they correct the earlier unconditional phase reset. + +To compare entry policies, hold the execution mode, session policy, model, +effort, tools, feedback and total budget fixed, and use separate trials: + +```yaml +kwargs: + execution_mode: heartbeat + task_entry: loopx-planned + planning_timeout_sec: 300 + iteration_context: fresh + turn_timeout_sec: 4700 + scheduler_timeout_sec: 5080 +``` + ## Install and isolate From the candidate worktree, provide: @@ -58,7 +105,7 @@ export PYTHONPATH="$PWD${PYTHONPATH:+:$PYTHONPATH}" Also set `CODEX_OFFLINE_DIR` (Codex, code-mode sidecar, rg), `LOOPX_PORTABLE_PYTHON` (Python >=3.11 distribution) and `LOOPX_NODE_DIR` -(Node >=22.18.0 distribution). Staging uses `git archive HEAD`, never local run +(Node >=22.18.0 distribution). Staging archives the verified commit SHA, never local run artifacts. The host import must come from that checkout, whose tracked files must match HEAD. Commit the candidate before real validation. Baselines stage only the runner/native transport, without installing LoopX skills or initializing diff --git a/docs/architecture/rfcs/long-horizon-harness-benchmark-research-program-v0.md b/docs/architecture/rfcs/long-horizon-harness-benchmark-research-program-v0.md index e6c0a55c83..5ab0dedd29 100644 --- a/docs/architecture/rfcs/long-horizon-harness-benchmark-research-program-v0.md +++ b/docs/architecture/rfcs/long-horizon-harness-benchmark-research-program-v0.md @@ -952,6 +952,14 @@ product heartbeat, Turn and Goal execution while retaining native verification. Synthetic Harbor conformance qualifies this engineering seam only; E3 matched studies and E4 cross-benchmark claims remain separate acceptance. +Task entry is a separate ablation axis: a runner-seeded execution Todo versus +model planning through the product's `todo plan` checkpoint. Planning runs +before the selected driver, uses the shared Goal planner/Todo-delta contract, +and consumes the phase budget without counting as advancement. Qualification +must read back task identity and actual Todos, preserve blocked state and +disclose the separate planning session; synthetic task success alone does not +prove equivalence to interactive `$loopx` startup or planning effectiveness. + ### 11.3 Required delivery slice Every benchmark engineering PR or contributor task should identify a bounded diff --git a/docs/architecture/rfcs/long-horizon-harness-benchmark-research-program-v0.zh-CN.md b/docs/architecture/rfcs/long-horizon-harness-benchmark-research-program-v0.zh-CN.md index 8fa26d4229..21083a51b2 100644 --- a/docs/architecture/rfcs/long-horizon-harness-benchmark-research-program-v0.zh-CN.md +++ b/docs/architecture/rfcs/long-horizon-harness-benchmark-research-program-v0.zh-CN.md @@ -809,6 +809,12 @@ LHTB/SWE-Marathon 原生桥接的工程检查点：共享 trial 初始化，复 heartbeat、Turn 和 Goal 执行，并保留原生验证。合成 Harbor conformance 只验证这条工程路径；E3 matched study 和 E4 跨 benchmark 结论仍需独立验收。 +任务入口是独立的消融轴：runner 预写执行 Todo，或模型通过产品 `todo plan` +检查点进行规划。规划先于所选执行驱动，复用 Goal planner 和 Todo 增量契约， +消耗 phase 总预算但不计作 advancement。验收须读回任务身份和真实 Todo，保留 +阻塞状态，并披露独立的规划会话；合成任务通过不能证明与交互式 `$loopx` 启动 +等价，也不能证明规划效果提升。 + ### 11.3 必需的 delivery slice 每个 benchmark engineering PR 或 contributor task 都应声明一个有界 slice，包含： diff --git a/docs/project-agent-todo-contract.md b/docs/project-agent-todo-contract.md index 227592edac..18f12cf246 100644 --- a/docs/project-agent-todo-contract.md +++ b/docs/project-agent-todo-contract.md @@ -29,6 +29,29 @@ sync catch up. ## Write Contract +For a caller-owned runtime that already registered its Goal/Agent, generate the +model planning checkpoint before writing executable task Todos: + +```bash +loopx --format json todo plan --goal-id --agent-id \ + --text '' +``` + +This read-only command reuses `/loopx`'s planner and Todo-delta contract. It +returns the current frontier, typed result schema and an explicit caller-owned +execution handoff; it does not run a model, create a Goal, write a Todo, activate +a host loop or spend quota. A model consumes the packet and uses the existing +Todo CLI to plan actual task work. No planning/setup Todo is required. Read back +the returned ids and current state before the caller activates its driver and +enters the normal quota guard. A planning result does not authorize execution. +Follow-up input preserves the Goal/Agent and existing waits; reconcile the plan +instead of restarting the Goal. Unrelated peers and their claims remain intact. + +The installed `$loopx` skill recognizes this explicit packet as a bounded +planning checkpoint. Without one, its existing startup/continuation behavior is +unchanged. Omit `todo plan` to use that ordinary interactive entry; callers must +not simulate a planning checkpoint by prewriting an advancement Todo. + When read-only analysis, a review packet, a gate checklist, or P0/P1 steering finds a concrete user or owner action, write it immediately with the todo CLI. Use `user_gate` only when the item blocks an agent or the whole goal: From db236f329041057fb586c77562fd9e3dd2c0575e Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Fri, 18 Sep 2026 02:46:00 +0800 Subject: [PATCH 4/7] fix(planning): exclude terminal gates from blocked readback Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- loopx/control_plane/goals/task_planning.py | 6 ++++-- tests/control_plane/test_task_planning.py | 20 ++++++++++++++++++++ 2 files changed, 24 insertions(+), 2 deletions(-) diff --git a/loopx/control_plane/goals/task_planning.py b/loopx/control_plane/goals/task_planning.py index fc76d68255..6a354b133e 100644 --- a/loopx/control_plane/goals/task_planning.py +++ b/loopx/control_plane/goals/task_planning.py @@ -16,6 +16,8 @@ from ..todos.todo_semantics import todo_item_is_actionable_open from ..todos.contract import ( TODO_STATUS_BLOCKED, + TODO_TERMINAL_STATUS_VALUES, + TODO_TASK_CLASS_ADVANCEMENT, TODO_TASK_CLASS_BLOCKER, TODO_TASK_CLASS_USER_GATE, ) @@ -73,7 +75,7 @@ def build_task_planning_packet( t for t in todos if t.get("role") == "agent" - and t.get("task_class") == "advancement_task" + and t.get("task_class") == TODO_TASK_CLASS_ADVANCEMENT and todo_item_is_actionable_open(t) ] fine = execution_profile_is_fine_grained(goal.get("execution_profile")) @@ -110,7 +112,7 @@ def build_task_planning_packet( "blocking_todo_ids": [ t["todo_id"] for t in todos - if t.get("status") != "done" + if t.get("status") not in TODO_TERMINAL_STATUS_VALUES and ( t.get("status") == TODO_STATUS_BLOCKED or t.get("task_class") diff --git a/tests/control_plane/test_task_planning.py b/tests/control_plane/test_task_planning.py index 2192c1e15e..826a78746a 100644 --- a/tests/control_plane/test_task_planning.py +++ b/tests/control_plane/test_task_planning.py @@ -91,6 +91,26 @@ def test_unknown_identity_rejected_without_creating_it(bound_goal): build_task_planning_packet(**(bound_goal | {"agent_id": "unknown"})) +def test_full_frontier_and_live_blockers_ignore_display_order_and_terminal_items(bound_goal): + state = bound_goal["project"] / "state.md" + work = [ + f"- [ ] [P0] Task {i}.\n" + f" " + for i in range(40) + ] + gates = [ + f"- [{'x' if status == 'done' else ' '}] [P0] Gate {status}.\n" + f" " + for status in ("open", "blocked", "done", "deferred") + ] + for ordered in (work, list(reversed(work))): + state.write_text("# Active Goal State\n\n## Agent Todos\n" + "\n".join(ordered) + + "\n\n## User Todos\n" + "\n".join(gates) + "\n") + packet = build_task_planning_packet(**bound_goal) + assert set(packet["runnable_todo_ids"]) == {f"todo_work_{i}" for i in range(40)} + assert set(packet["blocking_todo_ids"]) == {"todo_gate_open", "todo_gate_blocked"} + + def test_public_cli_returns_a_read_only_checkpoint_and_rejects_execution(bound_goal): command = [ "--format", From fc77656d4e4d25779f8f31f51e7c0776b45cfa78 Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Fri, 18 Sep 2026 02:54:16 +0800 Subject: [PATCH 5/7] fix(benchmark): reserve phase budget before every host wake Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- benchmark/runtime/RUNTIME.md | 4 ++++ benchmark/runtime/harbor.py | 8 ++++++-- benchmark/runtime/worker.py | 8 ++++++++ benchmark/tests/test_task_entry.py | 22 +++++++++++++++++++++- 4 files changed, 39 insertions(+), 3 deletions(-) diff --git a/benchmark/runtime/RUNTIME.md b/benchmark/runtime/RUNTIME.md index 689dab9c8e..fef6ea00b4 100644 --- a/benchmark/runtime/RUNTIME.md +++ b/benchmark/runtime/RUNTIME.md @@ -79,6 +79,10 @@ identity and expose existing Todos to the planner; they do not clear waiting state or force the agent active. An unresolved Turn must be recovered before another phase can replace its task input. These wait/recovery rules apply to both entry policies; they correct the earlier unconditional phase reset. +Every scheduler wake also checks the remaining phase deadline before opening a +host execution. If the configured execution window plus settlement reserve no +longer fits, it records a budget-exhausted no-op without creating a pending Turn. +The deadline uses the task environment's clock, including remote Harbor backends. To compare entry policies, hold the execution mode, session policy, model, effort, tools, feedback and total budget fixed, and use separate trials: diff --git a/benchmark/runtime/harbor.py b/benchmark/runtime/harbor.py index f50ac779a3..dd4699822c 100644 --- a/benchmark/runtime/harbor.py +++ b/benchmark/runtime/harbor.py @@ -598,10 +598,12 @@ async def run( if entry["status"] == "blocked": return remaining = int(deadline - time.monotonic()) - if remaining <= 150: + if remaining <= 160: raise TimeoutError("Task budget exhausted before execution handoff") # Planning consumes the phase budget, including when the host later resumes. - host_timeout = min(self.execution.timeout_seconds, remaining - 150) + # Keep ten seconds for scheduler startup before the worker checks + # its execution window plus the existing 150-second settlement reserve. + host_timeout = min(self.execution.timeout_seconds, remaining - 160) env["LOOPX_CODEX_TURN_TIMEOUT_SEC"] = str(host_timeout) if self.execution.mode in {"heartbeat", "turn"}: command = [ @@ -635,6 +637,8 @@ async def run( phase_log = f"/logs/agent/worker-phase-{self._phase_number:03d}.log" shell = ( "set +e; " + # Use the task environment's clock, including remote backends. + f"export LOOPX_PHASE_DEADLINE_EPOCH=$(( $(date +%s) + {remaining} )); " f"timeout --signal=TERM --kill-after=30 {remaining}s " f"{shlex.join(command)} >> {shlex.quote(phase_log)} 2>&1; " "rc=$?; " diff --git a/benchmark/runtime/worker.py b/benchmark/runtime/worker.py index f9f969f253..0bfad3010e 100644 --- a/benchmark/runtime/worker.py +++ b/benchmark/runtime/worker.py @@ -232,6 +232,14 @@ def run_once(env: dict[str, str]) -> dict: Path(env.get("LOOPX_RUNTIME_ROOT", str(home))) / "benchmark-pending-turn.json" ) try: + if stage == "execute" and env.get("LOOPX_PHASE_DEADLINE_EPOCH"): + remaining = float(env["LOOPX_PHASE_DEADLINE_EPOCH"]) - time.time() + # A later scheduler wake must still leave room for host execution, + # validation and settlement. Do not open a transaction the outer + # phase timeout would interrupt solely because it started too late. + if remaining <= execution.timeout_seconds + 150: + receipt.update(ok=True, budget_exhausted=True, host_invoked=False) + return receipt prepare_codex_home( home, execution=execution, diff --git a/benchmark/tests/test_task_entry.py b/benchmark/tests/test_task_entry.py index cf2501ca62..739cfb862d 100644 --- a/benchmark/tests/test_task_entry.py +++ b/benchmark/tests/test_task_entry.py @@ -259,7 +259,8 @@ async def read_result(**kwargs): if executions: command, env = executions[0] assert "--kill-after=30 300s" in command - assert float(env["LOOPX_CODEX_TURN_TIMEOUT_SEC"]) == 150 + assert float(env["LOOPX_CODEX_TURN_TIMEOUT_SEC"]) == 140 + assert "LOOPX_PHASE_DEADLINE_EPOCH=$(( $(date +%s) + 300 ))" in command def test_pending_turn_prevents_phase_input_replacement(tmp_path, monkeypatch): @@ -281,3 +282,22 @@ async def unexpected_write(*args, **kwargs): SimpleNamespace(exec=pending), "next task", cwd=str(tmp_path) ) ) + + +def test_late_scheduler_wake_does_not_open_an_unfinishable_turn(planning_env, monkeypatch): + from benchmark.runtime import worker + + env = planning_env | { + "LOOPX_EXECUTION_MODE": "turn", + "LOOPX_TASK_STAGE": "execute", + "LOOPX_VALIDATION_COMMAND_JSON": '["python", "check.py"]', + "LOOPX_CODEX_TURN_TIMEOUT_SEC": "60", + "LOOPX_PHASE_DEADLINE_EPOCH": "310", + } + monkeypatch.setattr(worker.time, "time", lambda: 100) + monkeypatch.setattr(worker, "prepare_codex_home", lambda *a, **kw: pytest.fail("late wake must not launch a host")) + for entry in ("seeded-todo", "loopx-planned"): + receipt = run_once(env | {"LOOPX_TASK_ENTRY": entry}) + assert receipt["budget_exhausted"] and receipt["host_invoked"] is False + assert receipt.get("turn_execution") is None + assert not (Path(env["LOOPX_RUNTIME_ROOT"]) / "benchmark-pending-turn.json").exists() From 458ae13596e1c10a8d44c412bdc3b64ae038b9e6 Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Fri, 18 Sep 2026 02:56:27 +0800 Subject: [PATCH 6/7] fix(benchmark): cap later wake windows by remaining phase time Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- benchmark/runtime/RUNTIME.md | 6 +++--- benchmark/runtime/worker.py | 11 +++++++---- benchmark/tests/test_task_entry.py | 22 +++++++++++++++++++++- 3 files changed, 31 insertions(+), 8 deletions(-) diff --git a/benchmark/runtime/RUNTIME.md b/benchmark/runtime/RUNTIME.md index fef6ea00b4..469ae564e6 100644 --- a/benchmark/runtime/RUNTIME.md +++ b/benchmark/runtime/RUNTIME.md @@ -79,9 +79,9 @@ identity and expose existing Todos to the planner; they do not clear waiting state or force the agent active. An unresolved Turn must be recovered before another phase can replace its task input. These wait/recovery rules apply to both entry policies; they correct the earlier unconditional phase reset. -Every scheduler wake also checks the remaining phase deadline before opening a -host execution. If the configured execution window plus settlement reserve no -longer fits, it records a budget-exhausted no-op without creating a pending Turn. +Every scheduler wake caps its host timeout against the remaining phase budget +before opening an execution. If only startup and settlement reserve remains, +it records a budget-exhausted no-op without creating a pending Turn. The deadline uses the task environment's clock, including remote Harbor backends. To compare entry policies, hold the execution mode, session policy, model, diff --git a/benchmark/runtime/worker.py b/benchmark/runtime/worker.py index 0bfad3010e..afd9fafd64 100644 --- a/benchmark/runtime/worker.py +++ b/benchmark/runtime/worker.py @@ -11,6 +11,7 @@ import time import uuid from contextlib import contextmanager +from dataclasses import replace from pathlib import Path from benchmark.runtime.codex import Execution, prepare_codex_home, process_environment @@ -234,12 +235,14 @@ def run_once(env: dict[str, str]) -> dict: try: if stage == "execute" and env.get("LOOPX_PHASE_DEADLINE_EPOCH"): remaining = float(env["LOOPX_PHASE_DEADLINE_EPOCH"]) - time.time() - # A later scheduler wake must still leave room for host execution, - # validation and settlement. Do not open a transaction the outer - # phase timeout would interrupt solely because it started too late. - if remaining <= execution.timeout_seconds + 150: + # Reserve startup and settlement on every wake, then allow the + # remaining time for work instead of reusing the initial timeout. + if remaining <= 160: receipt.update(ok=True, budget_exhausted=True, host_invoked=False) return receipt + execution = replace( + execution, timeout_seconds=min(execution.timeout_seconds, remaining - 160) + ) prepare_codex_home( home, execution=execution, diff --git a/benchmark/tests/test_task_entry.py b/benchmark/tests/test_task_entry.py index 739cfb862d..432f5bc6a3 100644 --- a/benchmark/tests/test_task_entry.py +++ b/benchmark/tests/test_task_entry.py @@ -292,7 +292,7 @@ def test_late_scheduler_wake_does_not_open_an_unfinishable_turn(planning_env, mo "LOOPX_TASK_STAGE": "execute", "LOOPX_VALIDATION_COMMAND_JSON": '["python", "check.py"]', "LOOPX_CODEX_TURN_TIMEOUT_SEC": "60", - "LOOPX_PHASE_DEADLINE_EPOCH": "310", + "LOOPX_PHASE_DEADLINE_EPOCH": "260", } monkeypatch.setattr(worker.time, "time", lambda: 100) monkeypatch.setattr(worker, "prepare_codex_home", lambda *a, **kw: pytest.fail("late wake must not launch a host")) @@ -301,3 +301,23 @@ def test_late_scheduler_wake_does_not_open_an_unfinishable_turn(planning_env, mo assert receipt["budget_exhausted"] and receipt["host_invoked"] is False assert receipt.get("turn_execution") is None assert not (Path(env["LOOPX_RUNTIME_ROOT"]) / "benchmark-pending-turn.json").exists() + + +def test_remaining_phase_time_caps_later_host_windows(planning_env, monkeypatch): + from benchmark.runtime import worker + + class CapturedWindow(Exception): + pass + + def capture(home, *, execution, **kwargs): + assert execution.timeout_seconds == 40 + raise CapturedWindow + + monkeypatch.setattr(worker.time, "time", lambda: 100) + monkeypatch.setattr(worker, "prepare_codex_home", capture) + with pytest.raises(CapturedWindow): + run_once(planning_env | { + "LOOPX_TASK_STAGE": "execute", + "LOOPX_CODEX_TURN_TIMEOUT_SEC": "60", + "LOOPX_PHASE_DEADLINE_EPOCH": "300", + }) From fe76e6caf25b5af5c4ab6449bfc6eec113896043 Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Fri, 18 Sep 2026 03:04:55 +0800 Subject: [PATCH 7/7] fix(benchmark): carry seeded task intent across native phases Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- benchmark/runtime/RUNTIME.md | 6 ++- benchmark/runtime/harbor.py | 35 +++++++++---- benchmark/tests/test_shared_codex_runtime.py | 2 +- benchmark/tests/test_task_entry.py | 52 ++++++++++++++++++++ 4 files changed, 83 insertions(+), 12 deletions(-) diff --git a/benchmark/runtime/RUNTIME.md b/benchmark/runtime/RUNTIME.md index 469ae564e6..682d04942c 100644 --- a/benchmark/runtime/RUNTIME.md +++ b/benchmark/runtime/RUNTIME.md @@ -51,8 +51,10 @@ validator protection remains the environment owner's responsibility. `task_entry` is independent of the execution mode: -- `seeded-todo` (the compatibility default) writes one generic execution Todo - per native phase. The agent can still plan and replan during execution. +- `seeded-todo` (the compatibility default) writes a generic execution Todo. + Follow-up phases update that Todo while it remains live and owned by this + agent; completed or deferred work gets a new Todo. Updates preserve blocked + state. The agent can still plan and replan during execution. - `loopx-planned` runs the installed `$loopx` skill against the public `loopx todo plan` checkpoint before execution. The checkpoint shares the product's planner and continuation-aware Todo delta; it creates no planning diff --git a/benchmark/runtime/harbor.py b/benchmark/runtime/harbor.py index dd4699822c..b2c7270114 100644 --- a/benchmark/runtime/harbor.py +++ b/benchmark/runtime/harbor.py @@ -82,6 +82,7 @@ def __init__( if not 1 <= self.replan_after_todos <= 5: raise ValueError("replan_after_todos must be between 1 and 5") self._phase_number = 0 + self._seeded_todo_id: str | None = None super().__init__(*args, **kwargs) @staticmethod @@ -385,8 +386,29 @@ async def _prepare_phase( ) async def _seed_phase(self, environment: BaseEnvironment, *, cwd: str) -> None: - todo_id = f"benchmark-task-phase-{self._phase_number:03d}" - await self._loopx( + text = ( + f"[P0] Execute benchmark phase {self._phase_number}. Read the exact " + f"current task from {self._task_document}; inspect the workspace, implement and " + "validate it, and create bounded successor Todos for remaining work." + ) + if self._seeded_todo_id: + listed = await self._loopx(environment, [ + "todo", "list", "--goal-id", _GOAL_ID, "--role", "agent", + "--todo-id", self._seeded_todo_id, + ], cwd=cwd) + current = next(iter(listed["todos"]), None) + if current and current.get("status") in {"open", "blocked"}: + if current.get("claimed_by") != _AGENT_ID: + raise RuntimeError("Seeded task Todo is no longer owned by this agent") + # New phase input revises our still-live generic task; do not + # strand it behind an unfinished predecessor or clear a wait. + await self._loopx(environment, [ + "todo", "update", "--goal-id", _GOAL_ID, + "--todo-id", self._seeded_todo_id, "--agent-id", _AGENT_ID, + "--text", text, "--execute", + ], cwd=cwd) + return + created = await self._loopx( environment, [ "todo", @@ -395,14 +417,8 @@ async def _seed_phase(self, environment: BaseEnvironment, *, cwd: str) -> None: _GOAL_ID, "--role", "agent", - "--todo-id", - todo_id, "--text", - ( - f"[P0] Execute benchmark phase {self._phase_number}. Read the exact " - f"current task from {self._task_document}; inspect the workspace, implement and " - "validate it, and create bounded successor Todos for remaining work." - ), + text, "--task-class", "advancement_task", "--action-kind", @@ -415,6 +431,7 @@ async def _seed_phase(self, environment: BaseEnvironment, *, cwd: str) -> None: ], cwd=cwd, ) + self._seeded_todo_id = created["todo_id"] def _worker_env(self, *, cwd: str) -> dict[str, str]: env = self._profile_env() diff --git a/benchmark/tests/test_shared_codex_runtime.py b/benchmark/tests/test_shared_codex_runtime.py index fcbcaa7e5b..144797ff55 100644 --- a/benchmark/tests/test_shared_codex_runtime.py +++ b/benchmark/tests/test_shared_codex_runtime.py @@ -357,7 +357,7 @@ async def cli(environment, args, **kwargs): # launching a model or mutating any active project. build_parser().parse_args(args) calls.append(args) - return {"after": {"execution_profile": {"replan_after_completed_todos": 3}}} + return {"todo_id": "todo_fixture", "after": {"execution_profile": {"replan_after_completed_todos": 3}}} monkeypatch.setattr(agent, "_write_task_document", write_task) monkeypatch.setattr(agent, "_registry_exists", registry_exists) diff --git a/benchmark/tests/test_task_entry.py b/benchmark/tests/test_task_entry.py index 432f5bc6a3..72b073b061 100644 --- a/benchmark/tests/test_task_entry.py +++ b/benchmark/tests/test_task_entry.py @@ -321,3 +321,55 @@ def capture(home, *, execution, **kwargs): "LOOPX_CODEX_TURN_TIMEOUT_SEC": "60", "LOOPX_PHASE_DEADLINE_EPOCH": "300", }) + + +@pytest.mark.parametrize("status", ["open", "blocked", "done", "deferred"]) +def test_seeded_followup_uses_real_todo_delta_without_reviving_terminal_work( + planning_env, tmp_path, monkeypatch, status +): + import contextlib + import io + pytest.importorskip("harbor") + from benchmark.runtime import harbor + from loopx.cli import main + + monkeypatch.setattr(harbor, "_GOAL_ID", "planning-goal") + monkeypatch.setattr(harbor, "_AGENT_ID", "planner") + agent = harbor.BenchmarkCodex(logs_dir=tmp_path, model_name="openai/fixture") + + async def cli(environment, args, **kwargs): + output = io.StringIO() + with contextlib.redirect_stdout(output): + code = main([ + "--format", "json", "--registry", planning_env["LOOPX_REGISTRY"], + "--runtime-root", planning_env["LOOPX_RUNTIME_ROOT"], *args, + ]) + assert code == 0, output.getvalue() + return json.loads(output.getvalue()) + + monkeypatch.setattr(agent, "_loopx", cli) + + async def scenario(): + agent._phase_number = 1 + await agent._seed_phase(None, cwd=planning_env["LOOPX_PROJECT"]) + original = agent._seeded_todo_id + transition = (["complete", "--no-follow-up", "--note", "Synthetic task independently validated; no remaining work"] + if status == "done" else ["update", "--status", status]) + if status == "deferred": + transition += ["--resume-when", "capacity_available:fixture_pool"] + await cli(None, ["todo", *transition, "--goal-id", "planning-goal", + "--todo-id", original, "--agent-id", "planner", "--execute"]) + agent._phase_number = 2 + await agent._seed_phase(None, cwd=planning_env["LOOPX_PROJECT"]) + listed = await cli(None, ["todo", "list", "--goal-id", "planning-goal", "--role", "agent"]) + todos = {t["todo_id"]: t for t in listed["todos"]} + if status in {"open", "blocked"}: + assert agent._seeded_todo_id == original and len(todos) == 1 + assert todos[original]["status"] == status + assert "task-phase-002.md" in todos[original]["text"] + else: + assert agent._seeded_todo_id != original and len(todos) == 2 + assert todos[original]["status"] == status + assert "task-phase-001.md" in todos[original]["text"] + + asyncio.run(scenario())