fix(ci): restore shared source qualification contracts

loopx/chat_configuration_api.py

@@ -2,7 +2,7 @@
 
 from collections.abc import Callable
 
-from . import chat_goal_ownership_api as ownership_api
+from .presentation import goal_ownership_api as ownership_api
 from . import chat_usage_statistics_api as usage_api
 from . import chat_goal_configuration_api as goal_api
 from . import chat_machine_configuration_api as machine_api

loopx/chat_goal_ownership_api.py → loopx/presentation/goal_ownership_api.py

@@ -11,7 +11,7 @@
 from urllib.parse import parse_qs, urlparse
 from uuid import uuid4
 
-from .control_plane.todos.provider_handoff_mode import (
+from ..control_plane.todos.provider_handoff_mode import (
     migrate_registered_handoff_mode,
     read_canonical_handoff_mode,
 )

tests/capabilities/test_capability_extension_registry.py

@@ -46,6 +46,7 @@
     "performance-diagnosis",
     "reliability-diagnostics",
     "progress-review-sentinel",
+    "goal-capability-organization",
 ]
 
 

tests/control_plane/test_cli_output_probe_runner.py

@@ -26,7 +26,18 @@ def turn_json_only(**kwargs):
             return {}
         return {"loopx_turn_plan": commands(**kwargs)["loopx_turn_plan"]}
 
+    def assert_crowded_turn_json_matrix(measurements):
+        # This alias test deliberately samples one JSON surface, whereas the
+        # production probe qualifies every surface/format and both scenarios.
+        assert set(measurements) == {"crowded"}
+        assert set(measurements["crowded"]) == {"loopx_turn_plan"}
+        formats = measurements["crowded"]["loopx_turn_plan"]
+        assert set(formats) == {"json"}
+        assert formats["json"]["json_parseable"] is True
+        assert formats["json"]["pretty_print_overhead_chars"] > 0
+
     monkeypatch.setattr(probe, "_surface_commands", turn_json_only)
+    monkeypatch.setattr(probe, "_assert_scenario_matrix", assert_crowded_turn_json_matrix)
     return runpy.run_path(str(RUNNER))["_default_rows"]
 
 

tests/control_plane/test_native_child_closeout_cli.py

@@ -8,16 +8,15 @@
 
 import pytest
 
-from test_native_child_replan_guard_cli import AGENT, GOAL, ROOT, TODO, TURN, _fixture
+from test_native_child_replan_guard_cli import AGENT, GOAL, ROOT, TODO, TURN, _admitted_guard, _fixture
 
 
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
 def test_closed_replan_only_accepts_existing_native_operations(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str,
 ) -> None:
     call, runtime, index = _fixture(tmp_path, monkeypatch, provider, True)
-    guard = call("quota", "should-run", "--codex-app", "--goal-id", GOAL,
-                 "--agent-id", AGENT, "--turn-instance-id", TURN)
+    guard = _admitted_guard(call, True)
     original = guard["heartbeat_receipt"]["settlement_identity"]
     base = ("native-child", "--goal-id", GOAL, "--agent-id", AGENT, "--turn-instance-id", TURN)
 

tests/control_plane/test_native_child_replan_guard_cli.py

@@ -78,26 +78,37 @@ def call(*args: str, expected_code: int = 0) -> dict:
     return call, runtime, index
 
 
+def _admitted_guard(call, todo_bound: bool) -> dict:
+    guard_args = ("quota", "should-run", "--codex-app", "--goal-id", GOAL,
+                  "--agent-id", AGENT, "--turn-instance-id", TURN)
+    guard = call(*guard_args)
+    if todo_bound:
+        # The planning recommendation has no settlement authority. An explicit
+        # choice may be retained during hard replan and bound only on reentry.
+        assert "settlement_identity" not in guard["heartbeat_receipt"]
+        rejected = call("native-child", "--goal-id", GOAL, "--agent-id", AGENT,
+            "--turn-instance-id", TURN, "record", "--operation-id", "op-before-choice",
+            "--stage", "decision", "--operation", "spawn", "--outcome", "started",
+            "--entrypoint-id", "generic_host", "--execute", expected_code=1)
+        assert "admitted" in rejected["error"]
+        deferred = call(*guard_args, "--todo-id", TODO, expected_code=1)
+        assert deferred["action_selection_qualification"]["state"] == "deferred"
+        assert "settlement_identity" not in deferred["heartbeat_receipt"]
+        [reentry] = deferred["interaction_contract"]["cli_channel"]["next_cli_actions"]
+        guard = call(*shlex.split(reentry)[1:])
+        assert guard["heartbeat_receipt"]["pending_action_selection"]["settlement_bound"] is True
+        assert guard["retained_action_selection"]["disposition"] == "preserve_retained_todo"
+    return guard
+
+
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
 @pytest.mark.parametrize("todo_bound", [True, False])
 def test_legal_replan_reports_native_child_without_settling_parent(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str, todo_bound: bool,
 ) -> None:
     call, runtime, index = _fixture(tmp_path, monkeypatch, provider, todo_bound)
-    guard = call("quota", "should-run", "--codex-app", "--goal-id", GOAL,
-                 "--agent-id", AGENT, "--turn-instance-id", TURN)
+    guard = _admitted_guard(call, todo_bound)
     assert guard["decision"] == "autonomous_replan_required", guard
-    if todo_bound:
-        # A recommendation does not bind the hard-replan Turn. Choose the
-        # existing Todo, then follow its retained-selection recovery command.
-        assert "settlement_identity" not in guard["heartbeat_receipt"]
-        assert guard["interaction_contract"]["cli_channel"]["selection_required"]
-        deferred = call("quota", "should-run", "--codex-app", "--goal-id", GOAL,
-                        "--agent-id", AGENT, "--turn-instance-id", TURN,
-                        "--todo-id", TODO, expected_code=1)
-        assert deferred["action_selection_qualification"]["state"] == "deferred"
-        [reentry] = deferred["interaction_contract"]["cli_channel"]["next_cli_actions"]
-        guard = call(*shlex.split(reentry)[1:])
     identity = guard["heartbeat_receipt"]["settlement_identity"]
     assert identity.get("todo_id") == (TODO if todo_bound else None)
     assert bool(identity.get("replan_obligation_id")) is not todo_bound