refactor(control-plane): render bootstrap steps through effect program

loopx/bootstrap_command_pack.py

@@ -13,6 +13,7 @@
     ISSUE_FIX_GOAL_CANDIDATE_DISCOVERY_COMMAND_TEMPLATE,
     build_issue_fix_goal_command_templates,
 )
+from .control_plane.effect_program import effect_program_from_ordered_steps
 from .host_loop_activation import (
     agent_type_for_host_surface,
     build_host_loop_activation_packet,
@@ -1787,48 +1788,47 @@ def actionable_shell_command(value: Any) -> str:
     commands = commands if isinstance(commands, dict) else {}
     selected_route = transaction.get("selected_capability_route")
     selected_route = selected_route if isinstance(selected_route, dict) else {}
-    steps = transaction.get("ordered_steps")
-    steps = steps if isinstance(steps, list) else []
+    ordered_steps = transaction.get("ordered_steps")
+    ordered_steps = ordered_steps if isinstance(ordered_steps, list) else []
+    program = effect_program_from_ordered_steps(ordered_steps)
     step_lines: list[str] = []
-    for index, step in enumerate(steps, start=1):
-        if not isinstance(step, dict):
-            continue
+    for index, step in enumerate(program.steps, start=1):
+        raw_step = step.raw
         command = (
-            step.get("command")
-            or step.get("command_template")
-            or step.get("command_source")
-            or step.get("prompt")
+            step.command
+            or raw_step.get("command_source")
+            or raw_step.get("prompt")
         )
-        if step.get("kind") == "capability_guard":
+        if step.kind == "capability_guard":
             entry_key = str(selected_route.get("entry_command_key") or "")
             admission_key = str(selected_route.get("admission_command_key") or "")
             entry_command = commands.get(entry_key)
             admission_command = commands.get(admission_key)
-            discovery_command = step.get("candidate_discovery_command_template")
+            discovery_command = raw_step.get("candidate_discovery_command_template")
             authority = (
                 "open public issue; source clues are evidence only"
-                if step.get("candidate_authority") == "public_open_tracker_issue"
-                else step.get("candidate_authority")
+                if raw_step.get("candidate_authority") == "public_open_tracker_issue"
+                else raw_step.get("candidate_authority")
             )
             step_lines.extend(
                 [
-                    f"{index}. `{step.get('id')}` ({step.get('kind')})",
+                    f"{index}. `{step.step_id}` ({step.kind})",
                     f"   - authority: {authority}",
                 ]
             )
             if discovery_command:
                 step_lines.append(
                     f"   - discover: `{str(discovery_command).splitlines()[0]}`"
                 )
-            preflight = step.get("candidate_preflight")
+            preflight = raw_step.get("candidate_preflight")
             if isinstance(preflight, dict):
                 required_evidence = " + ".join(
                     str(field)
                     for field in preflight.get("required_evidence_fields") or []
                 )
                 step_lines.append(
                     "   - preflight: refresh "
-                    f"{step.get('authority_refresh_required')}; "
+                    f"{raw_step.get('authority_refresh_required')}; "
                     f"provide {required_evidence}; "
                     f"{preflight.get('decision_rule')}"
                 )
@@ -1839,12 +1839,12 @@ def actionable_shell_command(value: Any) -> str:
                 ]
             )
             continue
-        step_label = f"{index}. `{step.get('id')}` ({step.get('kind')}): "
-        step_lines.append(step_label + str(step.get("purpose")))
+        step_label = f"{index}. `{step.step_id}` ({step.kind}): "
+        step_lines.append(step_label + str(step.purpose))
         if command:
             rendered_command = (
                 actionable_shell_command(command)
-                if step.get("id") == "connect_if_needed"
+                if step.step_id == "connect_if_needed"
                 else str(command).splitlines()[0]
             )
             step_lines.append(f"   - command/source: `{rendered_command}`")

loopx/control_plane/effect_program.py

@@ -57,6 +57,7 @@ class EffectStep:
     kind: str | None = None
     command: str | None = None
     purpose: str | None = None
+    raw: Mapping[str, Any] = field(default_factory=dict)
 
 
 @dataclass(frozen=True)
@@ -94,6 +95,7 @@ def effect_program_from_ordered_steps(
                     or None
                 ),
                 purpose=str(step.get("purpose") or "") or None,
+                raw=dict(step),
             )
         )
     return EffectProgram(

tests/control_plane/test_effect_program_ordered_steps.py

@@ -3,9 +3,47 @@
 import json
 from pathlib import Path
 
-from loopx.bootstrap_command_pack import build_start_goal_guided_packet
+from loopx.bootstrap_command_pack import (
+    build_start_goal_guided_packet,
+    render_start_goal_guided_markdown,
+)
 from loopx.control_plane.effect_program import effect_program_from_ordered_steps
 
+GOAL_ID = "guided-program-goal"
+AGENT_ID = "codex-guided-program"
+
+
+def _write_connected_project(tmp_path: Path) -> Path:
+    project = tmp_path / "project"
+    state_file = project / ".codex" / "goals" / GOAL_ID / "ACTIVE_GOAL_STATE.md"
+    state_file.parent.mkdir(parents=True)
+    state_file.write_text("# Active Goal State\n", encoding="utf-8")
+    registry = project / ".loopx" / "registry.json"
+    registry.parent.mkdir(parents=True)
+    registry.write_text(
+        json.dumps(
+            {
+                "schema_version": "0.1",
+                "goals": [
+                    {
+                        "id": GOAL_ID,
+                        "status": "active",
+                        "repo": str(project),
+                        "state_file": str(state_file.relative_to(project)),
+                        "coordination": {
+                            "agent_model": "peer_v1",
+                            "registered_agents": [AGENT_ID],
+                        },
+                    }
+                ],
+            }
+        )
+        + "\n",
+        encoding="utf-8",
+    )
+    return project
+
+
 
 def test_ordered_steps_map_to_effect_program() -> None:
     steps = [
@@ -58,40 +96,12 @@ def test_ordered_steps_ignore_non_mapping_entries() -> None:
 def test_real_bootstrap_ordered_steps_map_to_effect_program(
     tmp_path: Path,
 ) -> None:
-    project = tmp_path / "project"
-    state_file = (
-        project / ".codex" / "goals" / "guided-program-goal" / "ACTIVE_GOAL_STATE.md"
-    )
-    state_file.parent.mkdir(parents=True)
-    state_file.write_text("# Active Goal State\n", encoding="utf-8")
-    registry = project / ".loopx" / "registry.json"
-    registry.parent.mkdir(parents=True)
-    registry.write_text(
-        json.dumps(
-            {
-                "schema_version": "0.1",
-                "goals": [
-                    {
-                        "id": "guided-program-goal",
-                        "status": "active",
-                        "repo": str(project),
-                        "state_file": str(state_file.relative_to(project)),
-                        "coordination": {
-                            "agent_model": "peer_v1",
-                            "registered_agents": ["codex-guided-program"],
-                        },
-                    }
-                ],
-            }
-        )
-        + "\n",
-        encoding="utf-8",
-    )
+    project = _write_connected_project(tmp_path)
 
     payload = build_start_goal_guided_packet(
         project=project,
-        goal_id="guided-program-goal",
-        agent_id="codex-guided-program",
+        goal_id=GOAL_ID,
+        agent_id=AGENT_ID,
         cli_bin="loopx",
         host_surface="codex-app",
         goal_text="Ship a bounded public issue triage workflow.",
@@ -108,3 +118,25 @@ def test_real_bootstrap_ordered_steps_map_to_effect_program(
     assert program.steps
     assert program.steps[0].step_id == "inspect_connection"
     assert program.steps[0].kind == "read_only"
+
+
+def test_start_goal_guided_render_consumes_effect_program(
+    tmp_path: Path,
+) -> None:
+    project = _write_connected_project(tmp_path)
+    payload = build_start_goal_guided_packet(
+        project=project,
+        goal_id=GOAL_ID,
+        agent_id=AGENT_ID,
+        cli_bin="loopx",
+        host_surface="codex-app",
+        goal_text="Ship a bounded public issue triage workflow.",
+        available_capabilities=["network"],
+        include_command_pack_detail=False,
+    )
+
+    rendered = render_start_goal_guided_markdown(payload)
+
+    assert "inspect_connection" in rendered
+    assert "connect_if_needed" in rendered
+    assert "quota_guard" in rendered