diff --git a/docs/quota-allocation.md b/docs/quota-allocation.md index adcc5f3175..0b6cb7e3d9 100644 --- a/docs/quota-allocation.md +++ b/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. diff --git a/loopx/control_plane/quota/effect_program.py b/loopx/control_plane/quota/effect_program.py index c47571a36c..40acc8bcee 100644 --- a/loopx/control_plane/quota/effect_program.py +++ b/loopx/control_plane/quota/effect_program.py @@ -49,6 +49,7 @@ "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", @@ -56,11 +57,14 @@ "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: diff --git a/loopx/control_plane/quota/live_decision.py b/loopx/control_plane/quota/live_decision.py index c8af6e877a..23a98ebfab 100644 --- a/loopx/control_plane/quota/live_decision.py +++ b/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) diff --git a/tests/control_plane/test_scheduler_fallback_hint.py b/tests/control_plane/test_scheduler_fallback_hint.py index 22f7d4c5c4..2c20e5eeed 100644 --- a/tests/control_plane/test_scheduler_fallback_hint.py +++ b/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": {