fix(quota): resolve first-call replan selection before context delivery

docs/reference/protocols/turn-envelope-v0.md

@@ -921,17 +921,34 @@ read basis，同一 TS recommendation owner 仍在追加 run 前拒绝跨任务
 本项修复此前错误接受的跨任务写入，不清依赖、不改变 Todo 权威或授予新 Turn。
 
 For an eligible explicit choice deferred solely by runnable autonomous replan,
-the typed qualifier adds `inline_reentry_allowed=true`. After durably retaining
-that choice, the CLI now executes one same-Turn guard reentry and returns its
-fresh result directly, including in TurnEnvelope mode. This changes the previous
-default of returning a failed selection before asking the caller to reenter.
+the typed qualifier adds `inline_reentry_allowed=true`. The CLI executes one
+same-Turn guard reentry and returns its fresh result directly, including in
+TurnEnvelope mode. With an existing receipt it first durably retains the choice.
+On the first guard it carries that choice internally until the final receipt,
+without emitting an intermediate refusal or an unbound discovery packet. A
+concurrent receipt appearing during this first-call retry stops the retry;
+it cannot replace the other caller's binding. This changes the previous first-call
+refusal that required the model to read discovery and binding packets separately.
 It grants no delivery or spend authority: the fresh guard and receipt owner
 still decide the Todo/replan binding, and may return a newly changed gate. It
-does not retry other hard lanes, missing candidates, or a first-call refusal
-without a durable choice receipt. The reentry has no explicit selection, so it
+does not retry other hard lanes or missing candidates. The reentry has no explicit selection, so it
 cannot recursively retry. Hosts that do not execute this CLI path can continue
 to use the existing recovery command.
 
+When native context delivery is enabled, rejected guards keep their error,
+original receipt and recovery action within the configured output budget.
+The full captured rejection is optional diagnostic detail, available through
+the digest-checked reader. Oversized diagnostics use a selected read of those
+failure fields. Neither response enters pending context delivery or establishes
+a differential baseline; disabled context delivery keeps its ordinary output.
+
+中文：已知且合格的任务仅因自主重规划被推迟时，首次调用内部完成一次同 Turn
+重判，只交付最终绑定结果，不要求模型先读拒绝、再读发现包、再读绑定包。
+已有回执仍先持久化选择；首次重试期间出现并发回执则拒绝覆盖。
+重判仍检查当前停止条件和额度，后续交付仍需新 Turn。启用原生上下文传输时，
+拒绝响应有界交付错误、原回执及恢复动作；完整失败包仅供按需诊断，
+不会建立差分基线。关闭该功能时保留原输出。
+
 For other refusals, the agent receives
 `recovery_action=reenter_guard_without_selection` and one executable
 same-Turn guard in the full decision's `cli_channel.next_cli_actions`; the compact

loopx/cli_commands/quota.py

@@ -100,6 +100,7 @@
 )
 from .quota_failure_report import (
     QUOTA_EVENT_KINDS,
+    captured_quota_failure_view,
     quota_decision_transport_failure_payload,
     quota_failure_payload,
     quota_validation_failure_payload,
@@ -375,7 +376,13 @@ def _emit_quota_result(
             turn_id=_effective_spend_turn_instance_id(payload, heartbeat_turn_id=heartbeat_turn_id),
             started_at=usage_quota_started,
         )
-    context_delivery_enabled = payload.get("ok") is True and native_context_activation(args)["enabled"]
+    try:
+        context_delivery_enabled = native_context_activation(args)["enabled"]
+    except QuotaCommandValidationError:
+        if payload.get("ok") is True:
+            raise
+        # Preserve the original validation error when activation itself failed.
+        context_delivery_enabled = False
     bind_capture_selection_transport(payload, args)
     delivery_request = None
     decision_identity = ("", "")
@@ -386,6 +393,12 @@ def _emit_quota_result(
             "replan_read_argv": current_loopx_cli_argv()}
     try:
         capture_decision(capture_directory, payload)
+        if (payload.get("ok") is False and context_delivery_enabled and capture_directory
+                and args.format == "json" and not getattr(args, "defer_context_presentation", False)):
+            print_payload(captured_quota_failure_view(payload, capture_directory=capture_directory,
+                max_page_output_bytes=getattr(args, "max_page_output_bytes", None)),
+                args.format, _quota_renderer(args))
+            return 1
         if payload.get("ok") is True:
             delivery_request = handoff_decision(capture_directory, getattr(args, "decision_handoff_socket", None), context_delivery_enabled)
     except QuotaDecisionTransportError as exc:
@@ -539,6 +552,7 @@ def handle_quota_command(
     print_payload: PrintPayload,
     append_cli_rollout_event: RolloutEventAppender,
     capture_directory: Path | None = None,
+    retained_action_selection_todo_id: str | None = None,
 ) -> int:
     native_result = handle_native_context_command(args, print_payload=print_payload,
         registry_path=registry_path, runtime_root_arg=runtime_root_arg)
@@ -604,6 +618,7 @@ def handle_quota_command(
                 runtime_root=runtime_root,
                 turn_instance_id=heartbeat_turn_id,
                 goal_ref=goal_ref,
+                retained_todo_id=retained_action_selection_todo_id,
             )
             heartbeat_receipt_existing = action_selection.receipt
             payload = build_live_quota_should_run_decision(
@@ -667,6 +682,10 @@ def handle_quota_command(
                         print_payload=print_payload,
                         append_cli_rollout_event=append_cli_rollout_event,
                         capture_directory=capture_directory,
+                        retained_action_selection_todo_id=(
+                            action_selection.requested_todo_id
+                            if action_selection_preflight.receipt is None else None
+                        ),
                     )
                 heartbeat_receipt_existing = action_selection_preflight.receipt
                 heartbeat_receipt_existing_status = (

loopx/cli_commands/quota_action_selection.py

@@ -86,6 +86,7 @@ def load_requested_quota_action_selection(
     runtime_root: Path,
     turn_instance_id: str | None,
     goal_ref: Mapping[str, object] | None = None,
+    retained_todo_id: str | None = None,
 ) -> RequestedQuotaActionSelection:
     requested_todo_id = _requested_quota_action_todo_id(args)
     if not turn_instance_id:
@@ -104,13 +105,20 @@ def load_requested_quota_action_selection(
         turn_instance_id=turn_instance_id,
         goal_ref=goal_ref,
     )
+    if retained_todo_id is not None and existing:
+        # A first-call inline retry has not published a receipt. A concurrent
+        # caller acquiring this Turn must not be overwritten by its local choice.
+        raise HeartbeatReceiptIdentityConflictError(
+            "heartbeat receipt appeared during first-call selection reentry; "
+            "read the existing Turn receipt before retrying the original choice"
+        )
     if not existing:
         return RequestedQuotaActionSelection(
             requested_todo_id=requested_todo_id,
             receipt=None,
             receipt_bound_todo_id=None,
             receipt_bound_replan_obligation_id=None,
-            receipt_pending_action_todo_id=None,
+            receipt_pending_action_todo_id=retained_todo_id,
             receipt_identity_upgraded=False,
         )
     details_value = existing.get("details")
@@ -170,7 +178,7 @@ def _requested_quota_action_selection_preflight(
         agent_delivery_allowed=agent_delivery_allowed,
     )
     if receipt_bound_replan_obligation_id:
-        if not receipt_identity_upgraded:
+        if not receipt_identity_upgraded and not receipt_pending_action_todo_id:
             # A Turn that started directly in autonomous replan has no Todo
             # selection authority to replace.  A later same-Turn --todo-id is
             # therefore a harmless settled replay, not successor delivery.
@@ -298,18 +306,25 @@ def inline_action_selection_reentry_args(
     preflight: ActionSelectionPreflightResult,
     turn_instance_id: str | None,
 ) -> argparse.Namespace | None:
-    """Adapt one typed reentry instruction after the exact choice is durable."""
+    """Adapt one typed reentry, retaining the choice until the final receipt.
+
+    Before the first receipt there is no effect to recover: the caller can
+    carry its choice directly into one fresh guard. Existing receipts still
+    require durable retention before reentry.
+    """
     qualification = payload.get("action_selection_qualification")
     if (
         not isinstance(qualification, Mapping)
         or qualification.get("inline_reentry_allowed") is not True
-        or preflight.receipt is None
         or selection.requested_todo_id is None
         or turn_instance_id is None
-        or heartbeat_receipt_pending_action_todo_id(preflight.receipt)
-        != selection.requested_todo_id
     ):
         return None
+    if preflight.receipt is None:
+        if selection.receipt is not None:
+            return None
+    elif heartbeat_receipt_pending_action_todo_id(preflight.receipt) != selection.requested_todo_id:
+        return None
     # Re-enter the real guard; changed quota, gates, leases and capabilities
     # must be read again. No explicit selection on this hop means no recursion.
     reentry = argparse.Namespace(**vars(args))

loopx/cli_commands/quota_failure_report.py

@@ -9,6 +9,8 @@
 from __future__ import annotations
 
 import argparse
+import hashlib
+import json
 from collections.abc import Mapping
 from pathlib import Path
 
@@ -267,6 +269,59 @@ def quota_failure_payload(
     return payload
 
 
+def captured_quota_failure_view(
+    payload: Mapping[str, object], *, capture_directory: Path,
+    max_page_output_bytes: int | None = None,
+) -> dict[str, object]:
+    """Bound an opted-in refusal without entering native delivery continuity.
+
+    The rejected decision remains immutable diagnostic detail. Its current
+    error, receipt and recovery are useful input; unrelated Goal history is not
+    a prerequisite for following that recovery. No field is silently truncated.
+    """
+    from ..control_plane.turn_driver.decision_capture import (
+        context_capture_read_command, native_context_page_budget,
+    )
+
+    page_bytes, limit = native_context_page_budget()
+    if max_page_output_bytes is not None:
+        limit = min(limit, max_page_output_bytes)
+    saved = capture_directory / "decision.json"
+    digest = hashlib.sha256(saved.read_bytes()).hexdigest()
+    fields = ("ok", "goal_id", "decision", "should_run", "error_code", "reason",
+              "recommended_action", "heartbeat_receipt", "action_selection_conflict")
+    result = {key: payload[key] for key in fields if key in payload}
+    interaction = payload.get("interaction_contract")
+    cli = interaction.get("cli_channel") if isinstance(interaction, Mapping) else None
+    if isinstance(cli, Mapping) and "next_cli_actions" in cli:
+        result["next_cli_actions"] = cli["next_cli_actions"]
+    result["captured_decision"] = {"path": str(saved), "sha256": digest,
+        "command": context_capture_read_command(saved, digest,
+            page_bytes=min(page_bytes, limit), max_page_output_bytes=limit)}
+    result["instruction"] = (
+        "The guard refused this request. Follow its recovery using the original Turn identity. "
+        "The saved decision is optional diagnostic detail; read only facts needed for recovery. "
+        "This response does not establish a context baseline or grant execution authority."
+    )
+    def fits(value: dict[str, object]) -> bool:
+        return len(json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= limit
+
+    if not fits(result):
+        # Exceptional large diagnostics still use the shared digest-checked
+        # selected reader, never an eager replay of the full failed packet.
+        pointers = ["/" + key for key in fields if key in payload]
+        if isinstance(cli, Mapping) and "next_cli_actions" in cli:
+            pointers.append("/interaction_contract/cli_channel/next_cli_actions")
+        result = {"ok": False, "captured_decision": result["captured_decision"],
+            "instruction": "Read the selected failure and recovery fields before acting; "
+                "the original decision and Turn identity remain unchanged. No execution authority or context baseline was granted.",
+            "command": context_capture_read_command(saved, digest, select=pointers,
+                page_bytes=min(page_bytes, limit), max_page_output_bytes=limit)}
+        if not fits(result):
+            raise QuotaDecisionTransportError("Failure recovery metadata exceeds the output limit; read the saved original decision.")
+    return result
+
+
 def quota_decision_transport_failure_payload(
     args: argparse.Namespace,
     committed: Mapping[str, object],

tests/control_plane/test_quota_decision_capture.py

@@ -85,6 +85,72 @@ def test_existing_symlink_is_not_followed(tmp_path):
     assert not (tmp_path / 'absent').exists()
 
 
+@pytest.mark.parametrize('provider', ['file', 'sqlite'])
+def test_native_refusal_is_bounded_without_pending_context_or_admission(tmp_path, monkeypatch, provider):
+    import hashlib
+    from loopx.control_plane.turn_driver.codex_context import codex_context_hook
+    from tests.control_plane.test_quota_settlement_cli import (
+        _configure_selectable_alternative, _initialize_fixture_authority, _spend_run_count,
+    )
+
+    project, runtime, registry = _write_fixture(tmp_path)
+    _configure_selectable_alternative(project)
+    _initialize_fixture_authority(project, runtime, provider=provider)
+    capture, native = tmp_path / 'refused-capture', tmp_path / 'native-session'
+    native.mkdir(mode=0o700)
+    monkeypatch.setenv('CODEX_THREAD_ID', 'refusal-session')
+    monkeypatch.delenv('CODEX_ROLLOUT_TRACE_ROOT', raising=False)
+    codex_context_hook(str(native), dict(session_id='refusal-session', cwd=str(project),
+        hook_event_name='SessionStart', source='startup'))
+    journal = (native / 'session.json').read_bytes()
+    guard = ['quota', 'should-run', '--goal-id', GOAL_ID, '--agent-id', AGENT_ID,
+             '--todo-id', 'todo_missing', '--turn-instance-id', TURN_ID, '--codex-app']
+    rc, refused = _run_cli(registry, runtime, *guard, '--decision-output-dir', str(capture),
+        '--context-delivery', 'on', '--codex-context-dir', str(native), '--max-page-output-bytes', '4096')
+    raw = (capture / 'decision.json').read_bytes()
+    original = json.loads(raw)
+    assert rc == 1 and refused['ok'] is False
+    assert len(raw) > 4096
+    assert len(json.dumps(refused, ensure_ascii=False, separators=(',', ':')).encode()) + 1 <= 4096
+    assert refused['error_code'] == original['error_code'] == 'quota_action_selection_rejected'
+    assert refused['reason'] == original['reason']
+    assert refused['heartbeat_receipt'] == original['heartbeat_receipt']
+    assert refused['next_cli_actions'] == original['interaction_contract']['cli_channel']['next_cli_actions']
+    assert refused['captured_decision']['sha256'] == hashlib.sha256(raw).hexdigest()
+    assert (native / 'session.json').read_bytes() == journal
+    assert _heartbeat_receipt_count(runtime, TURN_ID) == _spend_run_count(runtime) == 0
+    # Explicitly disabled delivery preserves the ordinary failure view.
+    rc, disabled = _run_cli(registry, runtime, *guard, '--context-delivery', 'off')
+    assert rc == 1 and disabled['error_code'] == refused['error_code']
+    assert 'action_selection_qualification' in disabled and 'captured_decision' not in disabled
+
+
+def test_large_refusal_diagnostics_remain_reachable_by_selected_read(tmp_path):
+    from loopx.cli_commands.quota_failure_report import captured_quota_failure_view
+    from loopx.cli_commands.quota_capture import read_context_capture
+
+    payload = {'ok': False, 'reason': '必要的诊断证据' * 10000,
+               'heartbeat_receipt': {'status': 'not_committed'},
+               'unrelated_history': 'optional history ' * 10000}
+    capture_decision(tmp_path, payload)
+    result = captured_quota_failure_view(payload, capture_directory=tmp_path, max_page_output_bytes=4096)
+    assert result['ok'] is False and len(json.dumps(result).encode()) < 4096
+    from loopx.cli import build_parser
+
+    command = result['command']
+    content = []
+    while command:
+        tokens = shlex.split(command)
+        args = build_parser().parse_args(tokens[tokens.index('--format'):])
+        page = read_context_capture(args)
+        assert page['ok'], page
+        content.append(page['content'])
+        command = page.get('next_command')
+    selected = {item['pointer']: item['value'] for item in json.loads(''.join(content))['selections']}
+    assert selected['/reason'] == payload['reason']
+    assert '/unrelated_history' not in selected
+
+
 @pytest.mark.parametrize('provider', ['file', 'sqlite'])
 def test_real_cli_fallback_reference_keeps_capture_and_turn_effects(tmp_path, monkeypatch, provider):
     from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection