fix(turn): restore the dropped managed-step CLI surface

loopx/cli_commands/turn.py

@@ -45,7 +45,6 @@
 )
 from ..control_plane.turn_driver import (
     LOOPX_TURN_EXECUTION_SCHEMA_VERSION,
-    LOOPX_TURN_SESSION_BINDING_SCHEMA_VERSION,
     TurnRecoveryBlockedError,
     build_loopx_turn_command_validator,
     build_loopx_turn_plan,
@@ -68,10 +67,12 @@
 from .turn_dsh_host import build_dsh_host_runner
 from .turn_registration import register_turn_commands as register_turn_commands
 from .turn_inspection import handle_turn_journal_inspection
+from .turn_managed_step import handle_turn_managed_step
 from .turn_rendering import (
     render_loopx_turn_execution_markdown as _render_loopx_turn_execution_markdown,
     render_loopx_turn_plan_markdown as _render_loopx_turn_plan_markdown,
 )
+from .turn_selection import resolve_turn_resume_session_binding
 from .turn_todo_writeback import (
     write_turn_repair_update,
     write_turn_validated_completion,
@@ -109,6 +110,11 @@ def handle_turn_command(
     )
     if inspection_result is not None:
         return inspection_result
+    if args.turn_command == "managed-step":
+        return handle_turn_managed_step(
+            args, registry_path=registry_path, runtime_root_arg=runtime_root_arg,
+            output_format=output_format, print_payload=print_payload,
+        )
     try:
         scan_roots = [Path(item).expanduser() for item in args.scan_path]
         if not scan_roots:
@@ -187,35 +193,15 @@ def build_turn_decision(
         # so the advisory-primary rebinding lives in the shared decision owner
         # instead of being repeated per subcommand.
         decision = apply_controller_advisory_primary(build_turn_decision)
-        resume_identity = {
-            "goal_id": args.resume_goal_id,
-            "agent_id": args.resume_agent_id,
-            "todo_id": args.resume_todo_id,
-        }
-        supplied_resume_fields = [
-            field for field, value in resume_identity.items() if value is not None
-        ]
-        if supplied_resume_fields and len(supplied_resume_fields) != len(
-            resume_identity
-        ):
-            raise ValueError(
-                "resume planning requires --resume-goal-id, --resume-agent-id, "
-                "and --resume-todo-id together"
-            )
-        session_binding = None
-        if supplied_resume_fields:
-            session_binding = {
-                "schema_version": LOOPX_TURN_SESSION_BINDING_SCHEMA_VERSION,
-                **resume_identity,
-            }
+        resume_requested, session_binding = resolve_turn_resume_session_binding(args)
         turn_envelope = build_turn_envelope(
             decision,
             scheduler_execution_context=scheduler_context,
         )
         if (
             args.turn_command == "run-once"
             and args.host == "codex-cli"
-            and not supplied_resume_fields
+            and not resume_requested
             and turn_envelope.get("effective_action") != "governed_capability_intent"
         ):
             session_binding = codex_cli_session_binding(runtime_root, turn_envelope)
@@ -297,7 +283,7 @@ def build_turn_decision(
                     raise ValueError(
                         "--resume-turn-key cannot be combined with --turn-instance-id"
                     )
-                if supplied_resume_fields:
+                if resume_requested:
                     raise ValueError(
                         "--resume-turn-key cannot be combined with host session identity flags"
                     )

loopx/cli_commands/turn_registration.py

@@ -55,15 +55,16 @@ def register_turn_commands(
     # visible interactive mode would produce a default plan that cannot be
     # scheduled. The mode follows the *selected* host, never the environment.
     resolved_default_host = resolve_default_turn_host()
+    resolved_default_execution_mode = (
+        "isolated-headless"
+        if resolved_default_host == MANAGED_TURN_HOST
+        else "interactive-visible"
+    )
     _add_turn_decision_arguments(
         plan,
         default_host=resolved_default_host,
         host_choices=list(PLANNED_TURN_HOST_CHOICES),
-        default_execution_mode=(
-            "isolated-headless"
-            if resolved_default_host == MANAGED_TURN_HOST
-            else "interactive-visible"
-        ),
+        default_execution_mode=resolved_default_execution_mode,
     )
     plan.add_argument(
         "--include-transaction-detail",
@@ -83,6 +84,65 @@ def register_turn_commands(
     )
     plan.add_argument("--limit", type=int, default=5)
 
+    managed_step = command_sub.add_parser(
+        "managed-step",
+        help=(
+            "Decide one bounded same-Turn continuation for a failed Turn "
+            "without executing it."
+        ),
+        description=(
+            "Read one canonical Turn journal, rebuild its validated receipt, "
+            "and ask the pure Turn Loop Controller for a disposition against "
+            "the current decision. Grants no execution authority: it never "
+            "launches a host, writes state, or spends quota. The Turn journal "
+            "remains the authority for the attempt count and retry budget."
+        ),
+    )
+    add_subcommand_format(managed_step)
+    # A managed step decides a bounded headless continuation for a Turn that
+    # already failed, so it selects from the shipped run-once hosts and never
+    # plans a visible interactive mode.
+    _add_turn_decision_arguments(
+        managed_step,
+        default_host=resolved_default_host,
+        host_choices=list(RUN_ONCE_TURN_HOST_CHOICES),
+        execution_mode_choices=["isolated-headless"],
+        default_execution_mode="isolated-headless",
+    )
+    managed_step.add_argument(
+        "--turn-key",
+        required=True,
+        help="Exact sha256 Turn key of the failed Turn to decide about.",
+    )
+    managed_step.add_argument(
+        "--observed-attempt",
+        type=int,
+        help=(
+            "Caller's observed attempt count, reconciled against the Turn "
+            "journal. A disagreement is refused rather than adopted."
+        ),
+    )
+    managed_step.add_argument(
+        "--observed-max-attempts",
+        type=int,
+        help=(
+            "Caller's observed retry ceiling, reconciled against the Turn "
+            "journal retry policy."
+        ),
+    )
+    managed_step.add_argument(
+        "--scan-root",
+        default=default_public_scan_root(),
+        help="Public files to scan for obvious private material.",
+    )
+    managed_step.add_argument(
+        "--scan-path",
+        action="append",
+        default=[],
+        help="Specific public file or directory to scan. Repeatable.",
+    )
+    managed_step.add_argument("--limit", type=int, default=5)
+
     run_once = command_sub.add_parser(
         "run-once",
         help=(

loopx/cli_commands/turn_selection.py

@@ -3,6 +3,45 @@
 from collections.abc import Mapping
 from typing import Any
 
+from ..control_plane.turn_driver import LOOPX_TURN_SESSION_BINDING_SCHEMA_VERSION
+
+
+def resolve_turn_resume_session_binding(
+    args: Any,
+) -> tuple[bool, dict[str, Any] | None]:
+    """Resolve the explicit resume binding for one Turn as ``(requested, binding)``.
+
+    A resume names the exact Goal, Agent, and Todo session to continue. A partial
+    resume is refused rather than completed from ambient context, because the
+    binding is what the Turn journal fence compares against.
+
+    ``requested`` reports that the caller named a resume identity. That is not the
+    same fact as "a session binding exists": a run-once Codex CLI Turn derives one
+    from its envelope, and that derived binding must not be read back as an
+    explicit resume request.
+    """
+
+    identity = {
+        "goal_id": args.resume_goal_id,
+        "agent_id": args.resume_agent_id,
+        "todo_id": args.resume_todo_id,
+    }
+    supplied = [name for name, value in identity.items() if value is not None]
+    if not supplied:
+        return False, None
+    if len(supplied) != len(identity):
+        raise ValueError(
+            "resume planning requires --resume-goal-id, --resume-agent-id, "
+            "and --resume-todo-id together"
+        )
+    return (
+        True,
+        {
+            "schema_version": LOOPX_TURN_SESSION_BINDING_SCHEMA_VERSION,
+            **identity,
+        },
+    )
+
 
 def turn_controller_advisory_primary(
     decision: Mapping[str, Any],