fix(goal-start): fence guided bootstrap over orphaned Goal state

examples/bootstrap-command-pack-smoke.py

@@ -420,6 +420,128 @@ def test_start_goal_guided_previews_transaction_without_mutation() -> None:
         assert_fixture_unchanged(snapshot)
 
 
+def test_start_goal_guided_blocks_orphaned_goal_state() -> None:
+    """A reset that deleted the registry entry must not reopen the same goal."""
+
+    with tempfile.TemporaryDirectory() as tmp:
+        project = Path(tmp) / "reset-project"
+        project.mkdir()
+        goal_id = "reset-goal"
+        snapshot = write_connected_goal_fixture(
+            project, goal_id=goal_id, agent_id="codex-retired"
+        )
+        state_file = project / ".codex" / "goals" / goal_id / "ACTIVE_GOAL_STATE.md"
+        registry = project / ".loopx" / "registry.json"
+        registry.write_text(
+            json.dumps({"schema_version": "0.1", "goals": []}, indent=2) + "\n",
+            encoding="utf-8",
+        )
+
+        payload = run_json(
+            "start-goal",
+            "--guided",
+            "--project",
+            str(project),
+            "--goal-id",
+            goal_id,
+            "--host-surface",
+            "codex-app",
+            "--goal-text",
+            "Continue the interrupted refactor",
+        )
+
+        connection = payload["project_connection"]
+        assert connection["connection_state"] == "orphaned_goal_state", connection
+        assert connection["goal_found"] is False, connection
+        assert connection["bootstrap_continuation_allowed"] is False, connection
+        assert connection["orphaned_goal_state"]["state_file_routes"] == [
+            f".codex/goals/{goal_id}/ACTIVE_GOAL_STATE.md"
+        ], connection
+
+        transaction = payload["guided_transaction"]
+        assert transaction["blocked_by"] == "orphaned_goal_state", transaction
+        assert [step["id"] for step in transaction["ordered_steps"]] == [
+            "inspect_connection",
+            "resolve_orphaned_goal_state",
+        ], transaction
+
+        gate = transaction["orphaned_goal_state_gate"]
+        assert gate["schema_version"] == "loopx_orphaned_goal_state_gate_v0", gate
+        assert gate["forbidden_until_resolved"] == [
+            "bootstrap",
+            "agent_registration",
+            "todo_write",
+            "quota_spend",
+            "host_loop_activation",
+        ], gate
+        for route in gate["resolution_routes"]:
+            assert route["mutates"] is False, route
+            assert "--execute" not in route["command"], route
+            assert route["command"].splitlines()[-1].startswith("loopx "), route
+
+        commands = payload["command_pack"]["commands"]
+        for key in (
+            "goal_start_connect_if_needed",
+            "bootstrap_after_user_confirmation",
+            "goal_start_plan_prompt",
+        ):
+            assert commands[key] is None, key
+
+        safety = payload["safety_contract"]
+        assert safety["force_bootstrap_allowed"] is False, safety
+        assert safety["writes_state_file"] is False, safety
+        assert safety["orphaned_goal_state_blocks_continuation"] is True, safety
+        assert_packet_summary_refs(
+            payload,
+            packet_kind="guided_start_goal",
+            compact_projection_default=True,
+        )
+        assert_fixture_unchanged({registry: registry.read_text(), state_file: snapshot[state_file]})
+
+
+def test_start_goal_guided_blocks_orphaned_state_without_a_registry() -> None:
+    """The same invariant must hold when the reset deleted the whole registry file."""
+
+    with tempfile.TemporaryDirectory() as tmp:
+        project = Path(tmp) / "reset-project"
+        state_file = project / ".codex" / "goals" / "reset-goal" / "ACTIVE_GOAL_STATE.md"
+        state_file.parent.mkdir(parents=True)
+        state_text = "# Orphaned goal state written by a retired lane\n"
+        state_file.write_text(state_text, encoding="utf-8")
+
+        payload = run_json(
+            "start-goal",
+            "--guided",
+            "--project",
+            str(project),
+            "--goal-id",
+            "reset-goal",
+            "--host-surface",
+            "codex-app",
+            "--goal-text",
+            "Continue the interrupted refactor",
+        )
+
+        assert payload["project_connection"]["connection_state"] == "orphaned_goal_state"
+        transaction = payload["guided_transaction"]
+        assert transaction["blocked_by"] == "orphaned_goal_state", transaction
+        assert [step["id"] for step in transaction["ordered_steps"]] == [
+            "inspect_connection",
+            "resolve_orphaned_goal_state",
+        ], transaction
+        commands = payload["command_pack"]["commands"]
+        for key in (
+            "goal_start_connect_if_needed",
+            "goal_start_refresh_state",
+            "goal_start_host_loop_activation",
+            "goal_start_quota_should_run",
+            "goal_start_plan_prompt",
+        ):
+            assert commands[key] is None, key
+        assert state_file.read_text(encoding="utf-8") == state_text
+        assert not (project / ".loopx" / "registry.json").exists()
+
+
 def test_start_goal_guided_requires_explicit_goal_for_multi_goal_project() -> None:
     with tempfile.TemporaryDirectory() as tmp:
         project = Path(tmp) / "multi-goal-project"
@@ -816,6 +938,8 @@ def main() -> int:
     test_missing_project_stops_before_mutation()
     test_goal_text_invocation_plans_ranked_todos_before_activation()
     test_start_goal_guided_previews_transaction_without_mutation()
+    test_start_goal_guided_blocks_orphaned_goal_state()
+    test_start_goal_guided_blocks_orphaned_state_without_a_registry()
     test_start_goal_guided_requires_explicit_goal_for_multi_goal_project()
     test_connected_project_reuses_existing_state()
     test_linked_git_worktree_reuses_canonical_source_registry()

loopx/bootstrap_command_pack.py

@@ -14,6 +14,13 @@
     build_issue_fix_goal_command_templates,
 )
 from .control_plane.effect_program import effect_program_from_ordered_steps
+from .control_plane.goals.orphaned_goal_state import (
+    fence_command_pack,
+    guided_fence,
+    registry_missing_goal_connection,
+    render_guided_lines,
+    unregistered_goal_connection,
+)
 from .control_plane.goals.start_contract import (
     build_goal_start_contract,
     build_goal_start_prompt,
@@ -584,17 +591,12 @@ def inspect_bootstrap_connection(
         }
 
     if not registry:
-        return {
-            **base_connection,
-            "registry_exists": False,
-            "goal_id": inferred_goal_id,
-            "goal_found": False,
-            "state_file": str(state_file),
-            "state_file_exists": state_file.exists(),
-            "connection_state": "not_connected",
-            "mutation_confirmation_required": True,
-            "reason": "project-local .loopx/registry.json is missing",
-        }
+        return unregistered_goal_connection(
+            base_connection=base_connection,
+            project=resolved_project,
+            goal_id=inferred_goal_id,
+            state_file=state_file,
+        )
 
     goals = registry_goals(registry)
     selected_goal_id, selected_goal = _select_goal(goals, goal_id)
@@ -608,18 +610,14 @@ def inspect_bootstrap_connection(
     state_file = goal_state_file or fallback_state_file
 
     if selected_goal is None:
-        return {
-            **base_connection,
-            "registry_exists": True,
-            "goal_id": resolved_goal_id,
-            "goal_found": False,
-            "known_goal_ids": [str(goal.get("id")) for goal in goals],
-            "state_file": str(state_file),
-            "state_file_exists": state_file.exists(),
-            "connection_state": "registry_without_goal",
-            "mutation_confirmation_required": True,
-            "reason": "registry exists but no matching goal entry was found",
-        }
+        return registry_missing_goal_connection(
+            base_connection=base_connection,
+            project=resolved_project,
+            goal_id=resolved_goal_id,
+            known_goal_ids=[str(goal.get("id")) for goal in goals],
+            state_file=state_file,
+            registry_exists=True,
+        )
 
     if not selected_goal.get("state_file"):
         return {
@@ -1076,6 +1074,7 @@ def build_loopx_bootstrap_command_pack(
             "host_loop_activation_allowed": activation_allowed,
         },
     }
+    fence_command_pack(payload, command_prefix=command_prefix)
     if normalized_thread_id:
         payload["thread_id"] = normalized_thread_id
         payload["thread_agent_binding"] = thread_binding_projection
@@ -1709,6 +1708,10 @@ def rerun_start_goal(selected_agent_id: str) -> str:
             detail_command=detail_command,
         )
     )
+    orphaned_gate = command_pack.get("orphaned_goal_state")
+    if isinstance(orphaned_gate, dict):
+        guided_transaction.update(guided_fence(orphaned_gate))
+        guided_transaction.pop("identity_selection_gate", None)
     payload = {
         "ok": True,
         "schema_version": GUIDED_START_SCHEMA_VERSION,
@@ -1731,6 +1734,7 @@ def rerun_start_goal(selected_agent_id: str) -> str:
             "spends_quota": False,
             "mutation_commands_are_previewed": True,
             "force_bootstrap_allowed": False,
+            "orphaned_goal_state_blocks_continuation": isinstance(orphaned_gate, dict),
         },
     }
     if command_pack.get("thread_id"):
@@ -1887,6 +1891,7 @@ def actionable_shell_command(value: Any) -> str:
             + "\n".join(choices)
             + "\n"
         )
+    orphan_gate_lines = render_guided_lines(transaction)
     host_gate = transaction.get("host_surface_selection_gate")
     host_gate = host_gate if isinstance(host_gate, dict) else {}
     host_gate_lines = ""
@@ -1916,6 +1921,7 @@ def actionable_shell_command(value: Any) -> str:
 {chr(10).join(step_lines)}
 {host_gate_lines}
 {goal_gate_lines}
+{orphan_gate_lines}
 {identity_gate_lines}
 
 ## Todo Preservation