fix(planning): include requirement refinement documents

loopx/control_plane/goals/start_goal_todo_delta.py

@@ -140,18 +140,34 @@ 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 "<agent-id>"
+    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 '<agent-authored-file>' "
             "--expected-provider-revision '<current-provider-revision>' "
             "--operation-id '<stable-original-operation-id>' --execute"
         ),
+        "declaration_document_template": {
+            "outcome_id": "<outcome-id>",
+            "owner_agent_id": author,
+            "objective": "<outcome-objective>",
+            "requirements": [
+                {"id": "<requirement-id>", "description": "<requirement-description>"}
+            ],
+            "source": {"ref": "<source-ref>", "summary": "<source-summary>"},
+        },
+        "link_document_template": {
+            "outcome_id": "<same-outcome-id>",
+            "todo_ids": ["<actual-todo-id>"],
+        },
+        "revision_guidance": (
+            "Use provider_revision from current acceptance or inspect. Preserve owner and objective "
+            "when refining an existing outcome. After Todo writes, inspect again before linking. "
+            "Give each refinement its own stable operation id. Retry it unchanged after response loss."
+        ),
     }
     # This adapter already owns routed CLI templates for both planning entrypoints.
     readback = {
@@ -180,7 +196,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"
@@ -207,7 +223,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"
@@ -247,7 +263,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):

tests/control_plane/test_start_goal_compact_projection.py

@@ -1038,9 +1038,10 @@ def test_projection_materially_reduces_repetition_without_hiding_measurement(
     compact_json = json.dumps(compact, ensure_ascii=False, sort_keys=True)
     detailed_json = json.dumps(detailed, ensure_ascii=False, sort_keys=True)
 
-    # Exact-readback guidance is useful common content in both projections;
-    # retain the duplicate-content checks below rather than remove that guidance.
-    assert len(compact_json) <= int(len(detailed_json) * 0.66)
+    # Both views carry one copy of the declaration/link recipe. Its roughly
+    # 700 common characters contribute about 240 characters to the ratio
+    # difference (34% of the added content). Keep the duplicate limits below.
+    assert len(compact_json) <= int(len(detailed_json) * 0.66) + 240
     compact_duplication = compact["packet_summary"]["duplication_measurement"]
     detailed_duplication = detailed["packet_summary"]["duplication_measurement"]
     assert compact_duplication["objective_content"]["duplicate_occurrences"] <= 11

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 = {
+            "<agent-authored-file>": str(path),
+            "<current-provider-revision>": revision,
+            "<stable-original-operation-id>": 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"], {
+        "<outcome-id>": "repair",
+        "<outcome-objective>": bound_goal["text"],
+        "<requirement-id>": "stable-api",
+        "<requirement-description>": "Keep the API stable while repairing the failure.",
+        "<source-ref>": "user-task",
+        "<source-summary>": 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"], {
+        "<same-outcome-id>": "repair", "<actual-todo-id>": 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):