From 4217fd779a61d54ec873a6f7df19b95e702fc2f7 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 11 Nov 2026 09:15:13 +0800 Subject: [PATCH] fix(planning): include requirement refinement documents Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../goals/start_goal_todo_delta.py | 35 +++++-- tests/control_plane/test_task_planning.py | 92 ++++++++++++++----- 2 files changed, 93 insertions(+), 34 deletions(-) diff --git a/loopx/control_plane/goals/start_goal_todo_delta.py b/loopx/control_plane/goals/start_goal_todo_delta.py index 936b664546..f9dd509ee8 100644 --- a/loopx/control_plane/goals/start_goal_todo_delta.py +++ b/loopx/control_plane/goals/start_goal_todo_delta.py @@ -137,18 +137,35 @@ def todo_authoring_steps( if registry_path is not None: prefix += f" --registry {shell_arg(str(registry_path))}" acceptance = f"{prefix} --format json goal-acceptance" - route = f"--goal-id {shell_arg(str(goal_id))}" - if agent_id: - route += f" --agent-id {shell_arg(str(agent_id))}" - # The existing planning checkpoint owns declaration and later work linking. - # Keep their exact routed commands reachable without adding model stages. - requirement_commands = { + author = agent_id or "" + route = f"--goal-id {shell_arg(str(goal_id))} --agent-id {shell_arg(author)}" + # The planner supplies intent; goal-acceptance still validates each refinement. + requirement_authoring = { "inspect_command": f"{acceptance} inspect {route}", "refine_command_template": ( f"{acceptance} refine {route} --document '' " "--expected-provider-revision '' " "--operation-id '' --execute" ), + "declaration_document_template": { + "outcome_id": "", + "owner_agent_id": author, + "objective": "", + "requirements": [ + {"id": "", "description": ""} + ], + "source": {"ref": "", "summary": ""}, + }, + "link_document_template": { + "outcome_id": "", + "todo_ids": [""], + }, + "revision_guidance": ( + "Use the current provider_revision from the attached acceptance or inspect. " + "Keep the original owner and objective when adding requirements to an existing outcome. " + "After Todo writes, inspect for the current revision before linking. " + "Use a distinct stable operation id for each refinement and retry it unchanged after response loss." + ), } # This adapter already owns routed CLI templates for both planning entrypoints. readback = { @@ -177,7 +194,7 @@ def todo_authoring_steps( "id": "plan_ranked_todos", "kind": "model_checkpoint", "prompt": plan_prompt, - "goal_requirements": requirement_commands, + "goal_requirements": requirement_authoring, "purpose": ( "produce one public-safe, small, verifiable checkpoint Todo; keep " "later options as evidence-linked planning notes" @@ -204,7 +221,7 @@ def todo_authoring_steps( "id": "compare_planned_todos_with_frontier", "kind": "model_checkpoint", "prompt": plan_prompt, - "goal_requirements": requirement_commands, + "goal_requirements": requirement_authoring, "purpose": ( "compare the continuation with existing_runnable_frontier " "by text/action_kind before any writeback" @@ -244,7 +261,7 @@ def append_todo_delta_render_line( if isinstance(requirements, Mapping): step_lines.append( " - overall requirements: reuse existing or declare missing requirements before Todos, link actual ids, then read back; " - "use this checkpoint's goal_requirements commands in the JSON detail packet." + "use this checkpoint's goal_requirements commands and document templates in the JSON detail packet." ) todo_delta = raw_step.get("todo_delta") if not isinstance(todo_delta, Mapping): diff --git a/tests/control_plane/test_task_planning.py b/tests/control_plane/test_task_planning.py index d40952be6d..8df46edbb1 100644 --- a/tests/control_plane/test_task_planning.py +++ b/tests/control_plane/test_task_planning.py @@ -3,6 +3,7 @@ import contextlib import io import json +import shlex from pathlib import Path import pytest @@ -219,51 +220,92 @@ def test_public_cli_returns_a_read_only_checkpoint_and_rejects_execution(bound_g def test_first_plan_declares_then_links_and_reads_canonical_requirements(bound_goal): - def cli(*args): - out = io.StringIO() - with contextlib.redirect_stdout(out): - status = main(["--format", "json", "--registry", str(bound_goal["registry_path"]), - "--runtime-root", bound_goal["runtime_root_arg"], *args]) - result = json.loads(out.getvalue()) - assert status == 0, result + def cli(*args, expected_status=0): + output = io.StringIO() + with contextlib.redirect_stdout(output): + status = main(list(args)) + result = json.loads(output.getvalue()) + assert status == expected_status, result return result + prefix = [ + "--format", "json", "--registry", str(bound_goal["registry_path"]), + "--runtime-root", bound_goal["runtime_root_arg"], + ] route = ["--goal-id", "planning-goal", "--agent-id", "planner"] - before = build_task_planning_packet(**bound_goal) + plan_args = [ + *prefix, "todo", "plan", *route, "--project", + str(bound_goal["project"]), "--text", bound_goal["text"], + ] + before = cli(*plan_args) assert before["goal_acceptance"]["planning_requirements"]["linked_todo_ids"] == [] assert before["existing_todos"] == [] - document = {"outcome_id": "repair", "owner_agent_id": "planner", "objective": bound_goal["text"], - "requirements": [{"id": "stable-api", "description": "Keep the API stable while repairing the failure."}], - "source": {"ref": "user-task", "summary": bound_goal["text"]}} + authoring = before["ordered_steps"][0]["goal_requirements"] path = bound_goal["project"] / "requirements.json" + + def refine_args(revision, operation): + bindings = { + "": str(path), + "": revision, + "": operation, + } + program, *args = shlex.split(authoring["refine_command_template"]) + assert program == "loopx" + return [bindings.get(arg, arg) for arg in args] + + def fill(value, bindings): + if isinstance(value, dict): + return {key: fill(item, bindings) for key, item in value.items()} + if isinstance(value, list): + return [fill(item, bindings) for item in value] + return bindings.get(value, value) + + document = fill(authoring["declaration_document_template"], { + "": "repair", + "": bound_goal["text"], + "": "stable-api", + "": "Keep the API stable while repairing the failure.", + "": "user-task", + "": bound_goal["text"], + }) path.write_text(json.dumps(document)) - declaration = ["goal-acceptance", "refine", *route, "--document", str(path), - "--expected-provider-revision", before["goal_acceptance"]["provider_revision"], - "--operation-id", "first-declaration", "--execute"] + initial_revision = before["goal_acceptance"]["provider_revision"] + declaration = refine_args(initial_revision, "first-declaration") cli(*declaration) # The write committed, but its caller lost the response. declared = cli(*declaration) assert declared["status"] in {"replayed", "recovered"} outcome = declared["goal_acceptance_contract"]["outcome_requirements"]["outcomes"][0] assert outcome["todo_ids"] == [] - assert not build_task_planning_packet(**bound_goal)["existing_todos"], "Declaration precedes task Todos" - written = cli("todo", "add", "--goal-id", "planning-goal", "--role", "agent", "--claimed-by", "planner", - "--text", "[P0] Reproduce and repair the failure.", "--task-class", "advancement_task", - "--action-kind", "implement", "--execute") - current = cli("goal-acceptance", "inspect", *route) - path.write_text(json.dumps({"outcome_id": "repair", "todo_ids": [written["todo_id"]]})) - command = ["goal-acceptance", "refine", *route, "--document", str(path), - "--expected-provider-revision", current["provider_revision"], "--operation-id", "link-first-work", "--execute"] + assert not cli(*plan_args)["existing_todos"], "Declaration precedes task Todos" + written = cli( + *prefix, "todo", "add", "--goal-id", "planning-goal", "--role", "agent", + "--claimed-by", "planner", "--text", "[P0] Reproduce and repair the failure.", + "--task-class", "advancement_task", "--action-kind", "implement", "--execute", + ) + + link = fill(authoring["link_document_template"], { + "": "repair", "": written["todo_id"], + }) + path.write_text(json.dumps(link)) + stale = cli(*refine_args(initial_revision, "stale-link"), expected_status=1) + assert stale["ok"] is False + assert stale["error"] == "goal_acceptance_provider_revision_mismatch" + current = cli(*shlex.split(authoring["inspect_command"])[1:]) + command = refine_args(current["provider_revision"], "link-first-work") linked = cli(*command) replay = cli(*command) assert replay["status"] in {"replayed", "recovered"} assert replay["goal_acceptance_contract"] == linked["goal_acceptance_contract"] - after = build_task_planning_packet(**bound_goal) + after = cli(*plan_args) assert after["goal_acceptance"]["planning_requirements"] == { "agent_id": "planner", "declaration_current": True, "outcome_ids": ["repair"], "linked_todo_ids": [written["todo_id"]], } - assert after["goal_acceptance"]["goal_acceptance_contract"].get("criteria", []) == [], "Independent checks stay optional" - assert after["goal_acceptance"]["goal_acceptance_contract"]["outcome_requirements"]["outcomes"][0]["sources"] == [document["source"]] + contract = after["goal_acceptance"]["goal_acceptance_contract"] + assert contract.get("criteria", []) == [], "Independent checks stay optional" + outcome, = contract["outcome_requirements"]["outcomes"] + assert outcome["sources"] == [{"ref": "user-task", "summary": bound_goal["text"]}] + assert after["goal_waiting_on"] == before["goal_waiting_on"] def test_shared_planning_preserves_open_intent_and_stop_in_both_entries(bound_goal):