test(turn): keep the current-Turn decision owner guards on the owner

tests/test_loopx_turn_driver.py

@@ -1515,6 +1515,62 @@ def test_turn_cli_consumes_live_state_without_writes(
     assert before == after
 
 
+def test_turn_cli_resolves_its_decision_through_the_shared_owner(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """``turn`` must not re-inline the chain it shares with the managed step.
+
+    A private copy inside ``loopx/cli_commands/turn.py`` is not a formatting
+    problem: adding one decision input to the shared owner would then move only
+    one subcommand, and ``run-once`` and ``managed-step`` would disagree about
+    what the current Turn should do. Reading the live status through the shared
+    owner is the fact that identifies a private copy.
+    """
+
+    from loopx.cli_commands import turn_decision
+
+    project, runtime, registry = _write_live_fixture(tmp_path)
+    status_reads: list[dict[str, Any]] = []
+    real_collect_status = turn_decision.collect_status
+
+    def recording_collect_status(*args: Any, **kwargs: Any) -> dict[str, Any]:
+        status_reads.append(dict(kwargs))
+        return real_collect_status(*args, **kwargs)
+
+    monkeypatch.setattr(turn_decision, "collect_status", recording_collect_status)
+    output = io.StringIO()
+
+    with contextlib.redirect_stdout(output):
+        exit_code = cli_main(
+            [
+                "--registry",
+                str(registry),
+                "--runtime-root",
+                str(runtime),
+                "--format",
+                "json",
+                "turn",
+                "plan",
+                "--goal-id",
+                "loopx-turn-fixture",
+                "--agent-id",
+                "codex-fixture",
+                "--scan-root",
+                str(project),
+            ]
+        )
+
+    payload = json.loads(output.getvalue())
+    assert exit_code == 0, payload
+    assert payload["turn_envelope"]["schema_version"] == "loopx_turn_envelope_v0"
+    assert len(status_reads) == 1, (
+        "`turn` must read its live status through the shared Turn decision "
+        "owner so run-once and managed-step cannot drift; shared-owner status "
+        f"reads: {len(status_reads)}"
+    )
+
+
 def test_turn_cli_projects_explicit_fresh_iteration_context(
     tmp_path: Path,
 ) -> None:

tests/test_loopx_turn_journal_inspection.py

@@ -10,6 +10,7 @@
 
 from loopx.cli import main as cli_main
 from loopx.cli_commands import turn as turn_command
+from loopx.cli_commands import turn_decision
 from loopx.cli_commands import turn_rendering, turn_todo_writeback
 from loopx.control_plane.turn_driver import executor
 from loopx.control_plane.turn_driver import turn_journal_runtime
@@ -293,17 +294,26 @@ def unexpected_call(*args: object, **kwargs: object) -> None:
         raise AssertionError("inspect-journal reached a live or write path")
 
     for name in (
-        "build_lark_operator_inbox_urgency_projector",
-        "collect_status",
-        "scheduler_execution_context_for_turn",
         "build_live_quota_should_run_decision",
         "build_loopx_turn_plan",
+        "build_turn_envelope",
         "run_codex_cli_host",
         "run_loopx_turn_once",
         "spend_quota_slot",
         "refresh_state_run",
     ):
         monkeypatch.setattr(turn_command, name, unexpected_call)
+    # The shared decision owner now performs the live reads this command used to
+    # resolve itself, so the guard has to patch them where they live. Patching
+    # the old ``turn_command`` names would fail loudly here instead of proving
+    # anything: `turn` no longer resolves its own status, scheduler context or
+    # operator-inbox projector.
+    for name in (
+        "build_lark_operator_inbox_urgency_projector",
+        "collect_status",
+        "scheduler_execution_context_for_turn",
+    ):
+        monkeypatch.setattr(turn_decision, name, unexpected_call)
     for name in ("complete_goal_todo", "update_goal_todo"):
         monkeypatch.setattr(turn_todo_writeback, name, unexpected_call)
     monkeypatch.setattr(executor, "execute_turn_driver_settlement", unexpected_call)