fix(quota): bind the receipted Turn to the scheduler fallback hint

docs/quota-allocation.md

@@ -1690,8 +1690,13 @@ CLI-produced ACK hints bind that argument vector to the exact registry and
 effective runtime root that produced `quota should-run`. Hosts must execute the
 complete vector; stripping its leading global options can route a
 project-launched ACK into project-local scheduler state instead of the shared
-control plane. When the decision is heartbeat-receipted, ACK and failure hints
-also bind the originating `turn_instance_id`. The follow-up rebuilds the same
+control plane. `fallback_hint.cli_args` receives the same registry binding, and
+its `--turn-instance-id` value is always the reference the current host shell can
+expand: `"${LOOPX_TURN:?}"` on POSIX and `$LOOPX_TURN` on native Windows, so a
+PowerShell host does not replay an empty Turn. When the decision is
+heartbeat-receipted, ACK, failure, and fallback hints also bind the originating
+concrete `turn_instance_id`, and `fallback_hint.route_binding.turn_instance_bound`
+records whether that binding is present. The follow-up rebuilds the same
 receipt-bound live decision before validating its reset token and identity;
 later Todos or a different unscoped wait lane cannot replace the decision that
 requested the host action.

loopx/control_plane/quota/effect_program.py

@@ -49,18 +49,22 @@
     "SettlementStepKind",
     "build_codex_app_settlement_plan",
     "build_turn_scoped_cli_settlement_plan",
+    "is_turn_reference",
     "receipt_bound_monitor_phase",
     "receipt_bound_replay_phase",
     "receipt_bound_terminal_phase",
     "settlement_binding_args",
     "settlement_result_payload",
     "settlement_step_command",
     "turn_instance_arg",
+    "turn_reference",
 ]
 
 
 _POSIX_TURN_REF = "${LOOPX_TURN:?}"
 _POWERSHELL_TURN_REF = "$LOOPX_TURN"
+# Any POSIX shell reference names the same variable; only the guard differs.
+_POSIX_TURN_REFS = frozenset({_POSIX_TURN_REF, "${LOOPX_TURN}"})
 
 
 def _is_windows_host() -> bool:
@@ -79,6 +83,20 @@ def _quoted_turn_ref(turn_instance_id_ref: str) -> str:
     return shlex.quote(turn_instance_id_ref)
 
 
+def turn_reference(turn_instance_id: str | None = None) -> str:
+    """The concrete Turn when one is known, else the reference this host expands."""
+
+    return turn_instance_id or _default_turn_ref()
+
+
+def is_turn_reference(value: object) -> bool:
+    """Report whether a token is a shell-level Turn reference, not a concrete Turn."""
+
+    if not isinstance(value, str):
+        return False
+    return value in _POSIX_TURN_REFS or value == _POWERSHELL_TURN_REF
+
+
 def turn_instance_arg(turn_instance_id: str | None = None) -> str:
     """Render the ``--turn-instance-id`` argument for the shell that runs it.
 
@@ -87,8 +105,7 @@ def turn_instance_arg(turn_instance_id: str | None = None) -> str:
     native Windows host does not receive an empty Turn from POSIX syntax.
     """
 
-    reference = turn_instance_id or _default_turn_ref()
-    return f"--turn-instance-id {_quoted_turn_ref(reference)}"
+    return f"--turn-instance-id {_quoted_turn_ref(turn_reference(turn_instance_id))}"
 
 
 def _settlement_actor_args(arguments: str, agent_id: str) -> str:

loopx/control_plane/quota/live_decision.py

@@ -15,7 +15,11 @@
     InteractionProjectionHookRegistration,
     dispatch_interaction_projection_hooks,
 )
-from .effect_program import ReceiptBoundReplayPhase
+from .effect_program import (
+    ReceiptBoundReplayPhase,
+    is_turn_reference,
+    turn_reference,
+)
 from .blocked_retry import overlay_active_turn_retries
 from .settlement import (
     QuotaSettlementReadback,
@@ -380,6 +384,36 @@ def _apply_pending_capability_intent_precedence(
     return True
 
 
+def _bind_followup_turn_argument(
+    cli_args: list[Any],
+    reference: str,
+    *,
+    authoritative: bool,
+) -> str:
+    """Point a follow-up's ``--turn-instance-id`` value at ``reference`` in place.
+
+    An authoritative Turn always wins, because it owns the receipt the follow-up
+    replays. Without one only a shell reference is re-rendered, so a concrete
+    projected Turn survives untouched. Returns the value the vector now carries,
+    so a mirrored ``args`` block cannot disagree with the executed command.
+    """
+
+    index = (
+        cli_args.index("--turn-instance-id")
+        if "--turn-instance-id" in cli_args
+        else -1
+    )
+    if index < 0:
+        cli_args.extend(["--turn-instance-id", reference])
+        return reference
+    if index + 1 >= len(cli_args):
+        cli_args.append(reference)
+        return reference
+    if authoritative or is_turn_reference(cli_args[index + 1]):
+        cli_args[index + 1] = reference
+    return str(cli_args[index + 1])
+
+
 def bind_scheduler_followup_cli_routes(
     payload: dict[str, Any],
     *,
@@ -409,17 +443,33 @@ def bind_scheduler_followup_cli_routes(
             if hint_name == "fallback_hint":
                 if cli_args[0] != "loopx-apply-rrule" or "--registry" in cli_args:
                     continue
-                followup_hint["cli_args"] = [
+                bound_cli_args = [
                     cli_args[0],
                     "--registry",
                     str(registry_path.expanduser().resolve()),
                     *cli_args[1:],
                 ]
+                safe_turn_instance_id = str(turn_instance_id or "").strip()
+                # The fallback replays this decision through ``loopx quota
+                # should-run``, so it needs the Turn that owns the receipt.
+                # Projections leave a shell reference here; a native Windows
+                # host cannot expand the POSIX form and would replay an empty
+                # Turn instead of the decision it came from.
+                bound_turn = _bind_followup_turn_argument(
+                    bound_cli_args,
+                    turn_reference(safe_turn_instance_id),
+                    authoritative=bool(safe_turn_instance_id),
+                )
+                followup_hint["cli_args"] = bound_cli_args
+                args_value = followup_hint.get("args")
+                if isinstance(args_value, dict) and "turn_instance_id" in args_value:
+                    args_value["turn_instance_id"] = bound_turn
                 followup_hint["route_binding"] = {
                     "schema_version": "codex_app_scheduler_fallback_route_v0",
                     "source": source,
                     "registry_bound": True,
                     "runtime_root_bound": False,
+                    "turn_instance_bound": bool(safe_turn_instance_id),
                 }
                 continue
             bound_cli_args = list(cli_args)

tests/control_plane/test_scheduler_fallback_hint.py

@@ -203,6 +203,148 @@ def test_fallback_hint_cli_route_binds_registry(tmp_path: Path) -> None:
     assert fallback["route_binding"]["runtime_root_bound"] is False
 
 
+def _fallback_payload() -> dict:
+    return {
+        "scheduler_hint": {
+            "app_automation": {
+                "fallback_hint": {
+                    "cli_args": [
+                        "loopx-apply-rrule",
+                        "--goal-id",
+                        GOAL_ID,
+                        "--agent-id",
+                        AGENT_ID,
+                        "--automation-id",
+                        "loopx",
+                        "--turn-instance-id",
+                        "${LOOPX_TURN:?}",
+                    ],
+                    "args": {
+                        "goal_id": GOAL_ID,
+                        "agent_id": AGENT_ID,
+                        "automation_id": "loopx",
+                        "turn_instance_id": "${LOOPX_TURN:?}",
+                    },
+                }
+            }
+        }
+    }
+
+
+def test_fallback_hint_binds_the_receipted_turn(tmp_path: Path) -> None:
+    payload = _fallback_payload()
+
+    bind_scheduler_followup_cli_routes(
+        payload,
+        registry_path=tmp_path / "registry.json",
+        runtime_root=tmp_path / "runtime",
+        turn_instance_id="turn-fallback-receipt-001",
+    )
+
+    fallback = payload["scheduler_hint"]["app_automation"]["fallback_hint"]
+    turn_index = fallback["cli_args"].index("--turn-instance-id")
+    assert fallback["cli_args"][turn_index + 1] == "turn-fallback-receipt-001"
+    assert fallback["args"]["turn_instance_id"] == "turn-fallback-receipt-001"
+    assert fallback["route_binding"]["turn_instance_bound"] is True
+
+
+def test_fallback_hint_renders_the_turn_reference_of_the_host_shell(
+    tmp_path: Path,
+    monkeypatch,
+) -> None:
+    from loopx.control_plane.quota import effect_program as quota_effect_program
+
+    for windows_host, expected in (
+        (False, "${LOOPX_TURN:?}"),
+        (True, "$LOOPX_TURN"),
+    ):
+        monkeypatch.setattr(
+            quota_effect_program, "_is_windows_host", lambda value=windows_host: value
+        )
+        payload = _fallback_payload()
+
+        bind_scheduler_followup_cli_routes(
+            payload,
+            registry_path=tmp_path / "registry.json",
+            runtime_root=tmp_path / "runtime",
+        )
+
+        fallback = payload["scheduler_hint"]["app_automation"]["fallback_hint"]
+        turn_index = fallback["cli_args"].index("--turn-instance-id")
+        assert fallback["cli_args"][turn_index + 1] == expected
+        assert fallback["args"]["turn_instance_id"] == expected
+        assert fallback["route_binding"]["turn_instance_bound"] is False
+
+
+def test_fallback_hint_re_renders_every_posix_turn_reference(
+    tmp_path: Path,
+    monkeypatch,
+) -> None:
+    from loopx.control_plane.quota import effect_program as quota_effect_program
+
+    monkeypatch.setattr(quota_effect_program, "_is_windows_host", lambda: True)
+    for posix_form in ("${LOOPX_TURN:?}", "${LOOPX_TURN}"):
+        payload = _fallback_payload()
+        fallback = payload["scheduler_hint"]["app_automation"]["fallback_hint"]
+        turn_index = fallback["cli_args"].index("--turn-instance-id")
+        fallback["cli_args"][turn_index + 1] = posix_form
+        fallback["args"]["turn_instance_id"] = posix_form
+
+        bind_scheduler_followup_cli_routes(
+            payload,
+            registry_path=tmp_path / "registry.json",
+            runtime_root=tmp_path / "runtime",
+        )
+
+        bound_index = fallback["cli_args"].index("--turn-instance-id")
+        assert fallback["cli_args"][bound_index + 1] == "$LOOPX_TURN"
+        assert fallback["args"]["turn_instance_id"] == "$LOOPX_TURN"
+
+
+def test_fallback_hint_appends_a_missing_turn_argument(tmp_path: Path) -> None:
+    payload = _fallback_payload()
+    fallback = payload["scheduler_hint"]["app_automation"]["fallback_hint"]
+    fallback["cli_args"] = [
+        argument
+        for argument in fallback["cli_args"]
+        if argument != "--turn-instance-id"
+    ]
+    fallback["cli_args"].remove("${LOOPX_TURN:?}")
+    del fallback["args"]["turn_instance_id"]
+
+    bind_scheduler_followup_cli_routes(
+        payload,
+        registry_path=tmp_path / "registry.json",
+        runtime_root=tmp_path / "runtime",
+        turn_instance_id="turn-fallback-receipt-002",
+    )
+
+    assert fallback["cli_args"][-2:] == [
+        "--turn-instance-id",
+        "turn-fallback-receipt-002",
+    ]
+    assert fallback["route_binding"]["turn_instance_bound"] is True
+
+
+def test_fallback_hint_keeps_a_concrete_projected_turn(tmp_path: Path) -> None:
+    payload = _fallback_payload()
+    fallback = payload["scheduler_hint"]["app_automation"]["fallback_hint"]
+    turn_index = fallback["cli_args"].index("--turn-instance-id")
+    fallback["cli_args"][turn_index + 1] = "turn-projected-003"
+    fallback["args"]["turn_instance_id"] = "turn-projected-003"
+
+    bind_scheduler_followup_cli_routes(
+        payload,
+        registry_path=tmp_path / "registry.json",
+        runtime_root=tmp_path / "runtime",
+    )
+
+    bound_index = fallback["cli_args"].index("--turn-instance-id")
+    assert fallback["cli_args"][bound_index + 1] == "turn-projected-003"
+    assert fallback["args"]["turn_instance_id"] == "turn-projected-003"
+    assert fallback["route_binding"]["turn_instance_bound"] is False
+
+
 def test_scheduler_followup_routes_preserve_turn_lineage(tmp_path: Path) -> None:
     payload = {
         "scheduler_hint": {