refactor(quota): reference duplicate fallback diagnostics

docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md

@@ -576,7 +576,10 @@ changes no scheduler admission, execution grant, default or host effect.
 Real packaged transport qualification remains separate from live model outcomes,
 App adoption and the three convergence acceptances below.
 
-Remaining implementation Todos, in dependency order:
+The requirement-delivery (#5844), complete closeout-clause (#6337) and original
+Host decision-capture (#6350) source repairs have merged. Preserve those results;
+their completion does not close M7.4, R5 or the three acceptance streams below.
+Continue in this order using the existing owners and checkpoints:
 
 | Todo | Observable outcome and decisive acceptance |
 |---|---|
@@ -596,6 +599,28 @@ This closes the producer-to-detail transport gap only. Live models, packaged
 App/Lark, later authority checks and the remaining convergence matrix retain
 their independent acceptance; no Python projection is retired by this slice.
 
+The first ordinary-CLI consolidation replaces an exactly identical
+`heartbeat_recommendation.blocked_priority_fallback` mirror with
+`blocked_priority_fallback_ref="$.blocked_priority_fallback"`. The complete root
+fact, notification, execution obligation and original capture remain intact;
+different mirrors are retained, including Boolean/number differences. The existing
+`--include-detail agent-todos` view retains the legacy mirror. File/SQLite CLI
+tests cover same-capture action dimensions and no-debit replay. Original source
+hashes intentionally distinguish renderings; this does not change signature
+coverage or make a rendered view a substitute for the original capture.
+Legacy-field access stays in the existing packet owner; the CLI invokes its
+projection function without adding another legacy migration consumer.
+
+Next, inventory authorized real Goal captures across active work, peer/replan,
+wait/recovery and optional-capability shapes, then map each retained clause to
+its shared typed owner and actual consumer. Migrate one complete caller family
+before retiring its replaced projection. Compare the same captured decision
+through full/thin heartbeat and Turn, counting required detail reads, tokens,
+serialization/IO, latency and resource growth. Byte savings from this mirror
+removal qualify only that rendering delta; model quality, complete costs,
+installed App/Lark adoption and migration defaults still require their own
+evidence. Keep private source material out of public fixtures and checkpoints.
+
 The earlier cost slice reproduced a regression-budget gap, rather than a
 missing fixture alias or permission to remove required context. Comparing
 `aa87cc019` with `fc411c878` on the unchanged public CLI fixture, identical

docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md

@@ -394,7 +394,9 @@ TurnEnvelope 投影，不能只用未绑定的合成 argv。适配器只为递
 不改变 scheduler 准入、执行授权、默认值或宿主效果；打包后的传输验证不代表
 live 模型结果、App 采用或以下三项收敛验收完成。
 
-后续实施 Todo 按依赖顺序推进：
+完整要求交付（#5844）、完整收尾条款（#6337）和原始 Host 决策捕获（#6350）的
+源码修复已合入。保留这些成果；它们的完成不关闭 M7.4、R5 或下列三条验收线。
+后续沿既有 owner 和 checkpoint 按此顺序推进：
 
 | Todo | 可观察结果与决定性验收 |
 |---|---|
@@ -410,6 +412,22 @@ fixture。只读规划与可选 quota capture 默认值不变，原生执行 Tur
 这只关闭 producer 到 detail 的传输缺口。真实模型、打包 App/Lark、后续 authority 检查
 及其余收敛矩阵仍独立验收，本切片不退役任何 Python 投影。
 
+首个普通 CLI 整合将完全相同的
+`heartbeat_recommendation.blocked_priority_fallback` 副本替换为
+`blocked_priority_fallback_ref="$.blocked_priority_fallback"`。根字段的完整事实、
+通知、执行义务和原始 capture 保留；不同副本继续保留，包括 Boolean 与数字的差异。
+既有 `--include-detail agent-todos` 视图保留原副本。真实 File/SQLite CLI 覆盖同次
+捕获的动作维度及无扣费重放。原始 source hash 本就区分不同渲染；本批不改变签名
+覆盖，也不让渲染视图替代原始 capture。
+旧字段访问保留在既有 packet owner；CLI 调用其投影函数，不增加待迁移的旧字段消费者。
+
+下一步先调查获准真实 Goal 的活跃工作、peer/replan、等待/恢复和可选能力形态，
+将每条保留义务映射到共享 typed owner 与实际消费者，再完成一族调用方迁移，最后
+退役其旧投影。用同次捕获决策比较 full/thin heartbeat 与 Turn，将必需详情补读、
+token、序列化/IO、延迟及资源增长一并计入。此次移除副本的字节节省只证明该渲染
+差异；模型质量、完整成本、安装态 App/Lark 采用和迁移默认仍需独立证据。真实私有
+来源不进入公开 fixture 或 checkpoint。
+
 此前成本切片已复现回归预算缺口，不能归因于 fixture alias，也不授权删掉
 必需上下文。`aa87cc019` 与 `fc411c878` 使用相同公共 CLI fixture、临时 alias
 和命令参数，实际 JSON stdout 测量如下：

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

@@ -51,13 +51,29 @@ not verify file integrity or make saved admission current. Missing, malformed,
 or mismatched files require recovery, never automatic guard replay. Consumers already referring to `full_decision`, such as compacted capability
 context and peer diagnostics, now read this same observation. Scheduler-specific
 legacy detail requests are unchanged. Replan obligations and fresh selection/lease/quota checks retain their existing owners.
-Without capture, the historical full-decision route and all default outputs are
-unchanged. This opt-in detail link does not change action-signature coverage.
+Without capture, the historical full-decision route remains available. Enabling
+capture does not enable TurnEnvelope or change admission/settlement defaults.
+This opt-in detail link does not change action-signature coverage.
 
 中文：同时开启短包和完整 decision 保存时，详情入口读取本次保存的文件，不再
 为补读上下文重跑 guard。读取前核对身份和源哈希；旧观察不提供新的执行权限。
 文件损坏或丢失不能自动重跑，选 Todo、lease 变化等仍需按原契约重新准入。
 
+The default ordinary quota CLI view may replace an exactly identical nested
+`heartbeat_recommendation.blocked_priority_fallback` diagnostic with
+`blocked_priority_fallback_ref="$.blocked_priority_fallback"`. Read the complete
+root field in that same packet; resolving this reference needs no new guard or
+detail command. Different mirrors remain inline. Raw decisions, saved captures
+and the existing `--include-detail agent-todos` view retain their original shape.
+Action dimensions remain equivalent; the raw source hash still identifies the
+specific original rendering and must not be substituted across views.
+The existing packet owner supplies this compatibility projection, keeping the
+CLI from becoming another reader of the legacy guidance field.
+
+中文：普通 quota CLI 热路径将完全相同的 fallback 副本改为本包根字段引用，完整
+事实仍在当前包内，不需再次执行 guard。不同副本和既有详细视图保留；原始决定与
+capture 不变。动作维度等价不意味着不同渲染可以共用原始 source hash。
+
 ### Native Turn Host capture
 
 Executing `turn run-once --execute` now saves the original full decision before

loopx/control_plane/quota/cli_projection.py

@@ -3,6 +3,7 @@
 from typing import Any
 
 from .decision_summary import compact_quota_decision
+from .should_run_packet import reference_duplicate_fallback_guidance
 
 QUOTA_CLI_TODO_SUMMARY_COMPACTION_SCHEMA_VERSION = (
     "quota_cli_todo_summary_compaction_v0"
@@ -790,6 +791,7 @@ def compact_quota_should_run_cli_payload(
             compact = dict(compact)
             compact["goal_route_hint"] = _compact_goal_route_hint(goal_route_hint)
         compact = _compact_shadowed_action_projections(compact)
+        compact = reference_duplicate_fallback_guidance(compact)
     agent_lane_next_action = compact.get("agent_lane_next_action")
     if isinstance(agent_lane_next_action, dict) and "content_revision" in agent_lane_next_action:
         compact = dict(compact)

loopx/control_plane/quota/should_run_packet.py

@@ -1,4 +1,5 @@
 from __future__ import annotations
+import json
 from .effective_action import EffectiveAction
 from ..effect_runtime import effect_runtime_result
 from .effect_program import ReceiptBoundReplayPhase
@@ -152,6 +153,27 @@
 )
 
 
+def reference_duplicate_fallback_guidance(payload: dict[str, Any]) -> dict[str, Any]:
+    """Reference a CLI-only mirror through the existing legacy packet owner."""
+    fallback = payload.get("blocked_priority_fallback")
+    recommendation = payload.get("heartbeat_recommendation")
+    if not isinstance(fallback, dict) or not isinstance(recommendation, dict):
+        return payload
+    mirror = recommendation.get("blocked_priority_fallback")
+    if not isinstance(mirror, dict):
+        return payload
+    # Compare JSON types too: Python equality conflates False with 0.
+    if mirror is not fallback and (
+        json.dumps(mirror, sort_keys=True, ensure_ascii=False)
+        != json.dumps(fallback, sort_keys=True, ensure_ascii=False)
+    ):
+        return payload
+    recommendation = dict(recommendation)
+    recommendation.pop("blocked_priority_fallback")
+    recommendation["blocked_priority_fallback_ref"] = "$.blocked_priority_fallback"
+    return {**payload, "heartbeat_recommendation": recommendation}
+
+
 def _interaction_runtime_root(
     runtime_root: str | Path | None,
     status_payload: Mapping[str, Any],

tests/control_plane/test_quota_cli_projection.py

@@ -17,6 +17,37 @@
 )
 
 
+@pytest.mark.parametrize("mirror", ["shared", "identical", "different", "boolean-number", "missing-root"])
+def test_fallback_reference_preserves_complete_same_packet_facts(mirror: str) -> None:
+    fallback = {"notify_user": False, "blocked_items": [
+        {"text": "保留完整原因和恢复条件。" * 200, "reason": "Wait for the original dependency"},
+    ], "selected_executable": {"todo_id": "todo_fallback", "claimed_by": "worker"}}
+    nested = fallback if mirror == "shared" else deepcopy(fallback)
+    if mirror == "different":
+        nested["blocked_items"][0]["reason"] = "A different dependency"
+    elif mirror == "boolean-number":
+        nested["notify_user"] = 0
+    payload = {"heartbeat_recommendation": {"notify": "DONT_NOTIFY", "agent_must_attempt": True,
+                                             "blocked_priority_fallback": nested}}
+    if mirror != "missing-root":
+        payload["blocked_priority_fallback"] = fallback
+    original = deepcopy(payload)
+    compact = compact_quota_should_run_cli_payload(payload)
+    recommendation = compact["heartbeat_recommendation"]
+    if mirror in {"shared", "identical"}:
+        assert "blocked_priority_fallback" not in recommendation
+        assert recommendation["blocked_priority_fallback_ref"] == "$.blocked_priority_fallback"
+        assert compact["blocked_priority_fallback"] == fallback
+        assert recommendation["agent_must_attempt"] is True
+        assert recommendation["notify"] == "DONT_NOTIFY"
+        assert len(json.dumps(compact, ensure_ascii=False)) < len(json.dumps(original, ensure_ascii=False))
+    else:
+        assert recommendation["blocked_priority_fallback"] == nested
+        assert "blocked_priority_fallback_ref" not in recommendation
+    assert payload == original
+    assert compact_quota_should_run_cli_payload(payload, include_todo_summary_detail=True) == original
+
+
 @pytest.mark.parametrize("inline", ["absent", "identical", "different", "wrong-step"])
 def test_replan_authoring_is_available_without_another_read(inline: str) -> None:
     authoring = {"schema_version": "goal_vision_replan_contract_v0",

tests/control_plane/test_quota_decision_capture.py

@@ -67,6 +67,56 @@ def test_existing_symlink_is_not_followed(tmp_path):
     assert not (tmp_path / 'absent').exists()
 
 
+@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
+    from loopx.control_plane.quota.turn_envelope import quota_action_signature_document
+    from tests.control_plane.canonical_authority_fixture import (
+        initialize_canonical_authority, isolate_sqlite_runtime,
+    )
+    from tests.control_plane.test_quota_settlement_cli import _spend_run_count
+
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    project, runtime, registry = _write_fixture(tmp_path)
+    state = _configure_read_only_todo(project)
+    with state.open('a') as stream:
+        stream.write('\n- [ ] [P0] Preserve the original dependency and recovery evidence.\n'
+                     f'  <!-- loopx:todo todo_id=todo_fixture_blocked status=blocked '
+                     f'claimed_by={AGENT_ID} task_class=advancement_task '
+                     'action_kind=validate reason=fixture_dependency -->\n')
+    code, listed = _run_cli(registry, runtime, 'todo', 'list', '--goal-id', GOAL_ID)
+    assert code == 0, listed
+    projection = build_todo_runtime_shadow_projection(
+        goal_id=GOAL_ID, handoff_mode='soft_claim', todos=listed['todos'],
+    )
+    initialize_canonical_authority(runtime, GOAL_ID, projection, state_path=state, provider=provider)
+    args = ['quota', 'should-run', '--goal-id', GOAL_ID, '--agent-id', AGENT_ID,
+            '--todo-id', TODO_ID, '--turn-instance-id', TURN_ID, '--codex-app',
+            '--available-capability', 'shell', '--available-capability', 'filesystem_read']
+    capture = tmp_path / 'original-decision'
+    code, ordinary = _run_cli(registry, runtime, *args, '--decision-output-dir', str(capture))
+    assert code == 0, ordinary
+    saved = capture / 'decision.json'
+    original_bytes = saved.read_bytes()
+    full = json.loads(original_bytes)
+    fallback = full['blocked_priority_fallback']
+    assert full['heartbeat_recommendation']['blocked_priority_fallback'] == fallback
+    assert ordinary['blocked_priority_fallback'] == fallback
+    assert ordinary['heartbeat_recommendation']['blocked_priority_fallback_ref'] == '$.blocked_priority_fallback'
+    assert 'blocked_priority_fallback' not in ordinary['heartbeat_recommendation']
+    assert ordinary['heartbeat_receipt'] == full['heartbeat_receipt']
+    assert ordinary['interaction_contract']['cli_channel']['settlement_plan'] == full[
+        'interaction_contract']['cli_channel']['settlement_plan']
+    # Raw source hashes distinguish renderings; every action dimension must match.
+    assert quota_action_signature_document(ordinary) == quota_action_signature_document(full)
+    before = _heartbeat_receipt_count(runtime, TURN_ID)
+    code, replay = _run_cli(registry, runtime, *args)
+    assert code == 0, replay
+    assert _heartbeat_receipt_count(runtime, TURN_ID) == before
+    assert _spend_run_count(runtime) == 0
+    assert saved.read_bytes() == original_bytes
+
+
 @pytest.mark.parametrize('provider', ['file', 'sqlite'])
 def test_real_cli_boundary_clauses_reach_shared_host_without_loss(tmp_path, monkeypatch, provider):
     from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection