fix(quota): reference repeated full task titles in CLI packets

docs/reference/protocols/quota-cli-hot-path-compaction-v0.md

@@ -67,6 +67,24 @@ projection. Candidate lists and peer action lists retain counts and point to
 `--include-detail vision`; `--include-detail all` restores every supported
 detail section.
 
+When a bounded next-action label still carries the complete Task again in
+`agent_lane_next_action.title`, the default view may replace that exact duplicate
+with `title_ref`. The reference resolves to `text` or `title` in the single
+current selected-Todo source under this response's `work_context.sources`.
+The typed work-context owner must have verified the full current requirements;
+incomplete, failed, ambiguous, mismatched or differing sources retain the title.
+The full requirements, source metadata, controls and original capture remain
+intact. `--include-detail agent-todos` retains the title inline. This ordinary
+CLI serialization applies with differential delivery off as well as on; it
+neither enables native tracing nor changes the default-off transport switch.
+
+下一动作的有界摘要可能在 `agent_lane_next_action.title` 中重复携带完整任务。
+默认视图仅将与当前包内唯一、完整、已核验 Todo 源精确相同的副本替换为
+`title_ref`，指向 `work_context.sources` 中的 `text` 或 `title`。
+来源缺失、不完整、失败、歧义、身份不符或内容不同时保留原标题；完整要求、来源、
+控制字段和原始捕获均保留。`--include-detail agent-todos` 保留内联标题。
+这是普通 CLI 序列化去重，差分开关关闭和开启时均适用，不会开启原生追踪或改变默认关闭。
+
 When a replan action carries a complete `vision_authoring` schema, the default
 `quota should-run` packet now retains it, alongside its executable writeback
 summary (`required_fields`, accepted path outcomes, and rule). Only when the

loopx/control_plane/quota/cli_projection.py

@@ -582,18 +582,59 @@ def _deduplicate_next_action_warning(
     return compact
 
 
-def _deduplicate_agent_lane_next_action(item: dict[str, Any]) -> dict[str, Any]:
+def _deduplicate_agent_lane_next_action(
+    item: dict[str, Any], *, payload: dict[str, Any],
+) -> dict[str, Any]:
     text = item.get("text")
     title = item.get("title")
-    if (
-        not isinstance(text, str)
-        or not isinstance(title, str)
-        or (text != title and not text.endswith(f" {title}"))
-    ):
+    if not isinstance(title, str):
         return item
-    compact = dict(item)
-    compact.pop("title", None)
-    return compact
+    if isinstance(text, str) and (text == title or text.endswith(f" {title}")):
+        compact = dict(item)
+        compact.pop("title", None)
+        return compact
+
+    # The typed work-context owner has already verified the current full Task.
+    # A bounded lane label may differ from its full title. Reference only an
+    # exact duplicate in this same packet, never a summary or a prior capture.
+    contract = payload.get("interaction_contract")
+    channel = contract.get("agent_channel") if isinstance(contract, dict) else None
+    context = channel.get("work_context") if isinstance(channel, dict) else None
+    selected = payload.get("selected_todo")
+    if not isinstance(context, dict) or not isinstance(selected, dict):
+        return item
+    if (context.get("complete") is not True or context.get("selected_todo_text_verified") is not True
+            or not isinstance(item.get("todo_id"), str) or not item["todo_id"] or "title_ref" in item
+            or selected.get("todo_id") != item["todo_id"]):
+        return item
+    sources = context.get("sources", [])
+    if not isinstance(sources, list):
+        return item
+    matches = [(index, source.get("content", {})) for index, source in enumerate(sources)
+        if isinstance(source, dict) and source.get("source") == "selected_todo"]
+    if len(matches) != 1:
+        return item
+    index, content = matches[0]
+    if not isinstance(content, dict):
+        return item
+    todo = content.get("todo", {})
+    if (not isinstance(todo, dict) or not isinstance(todo.get("text"), str)
+            or content.get("ok") is not True or content.get("matched") is not True
+            or content.get("ambiguous") is True or content.get("read_complete") is False
+            or todo.get("archive_state") == "archive" or todo.get("status") == "done"
+            or todo.get("todo_id") != item["todo_id"]):
+        return item
+    for field in ("text", "title"):
+        if todo.get(field) != title:
+            continue
+        ref = f"$.interaction_contract.agent_channel.work_context.sources[{index}].content.todo.{field}"
+        if len(json.dumps(title, ensure_ascii=False)) <= len(ref) + 8:
+            return item
+        compact = dict(item)
+        compact.pop("title")
+        compact["title_ref"] = ref
+        return compact
+    return item
 
 
 def _compact_goal_route_hint(route: dict[str, Any]) -> dict[str, Any]:
@@ -885,7 +926,7 @@ def compact_quota_should_run_cli_payload(
         if isinstance(agent_lane_next_action, dict):
             compact = dict(compact)
             compact["agent_lane_next_action"] = _deduplicate_agent_lane_next_action(
-                agent_lane_next_action
+                agent_lane_next_action, payload=compact,
             )
         goal_route_hint = payload.get("goal_route_hint")
         if isinstance(goal_route_hint, dict):

tests/control_plane/test_quota_cli_projection.py

@@ -17,6 +17,105 @@
 )
 
 
+@pytest.mark.parametrize("variant", ["text", "title", "different", "short", "incomplete", "unverified",
+    "unmatched", "ambiguous", "partial", "failed", "wrong-id", "wrong-selection", "duplicate", "existing-ref",
+    "archived", "done", "boolean-id"])
+def test_lane_title_reuses_only_exact_complete_current_task_in_same_packet(variant):
+    body = "Preserve every requirement and source. 完整要求🙂\n" * 100 + "Stop before publication."
+    title = body if variant != "short" else "Short label"
+    todo = {"todo_id": "todo_current", "text": body, "status": "open"}
+    if variant == "title":
+        todo.update(text="[P1] " + body, title=body)
+    if variant == "different":
+        todo["text"] = body + " Additional acceptance."
+    content = {"ok": True, "matched": True, "ambiguous": False, "todo": todo}
+    context = {"complete": True, "selected_todo_text_verified": True,
+        "sources": [{"source": "goal_acceptance", "content": {}},
+                    {"source": "selected_todo", "content": content}]}
+    lane = {"todo_id": "todo_current", "text": "Bounded display…", "title": title,
+        "claimed_by": "worker", "required_capabilities": ["shell"]}
+    payload = {"ok": True, "selected_todo": {"todo_id": "todo_current"},
+        "agent_lane_next_action": lane, "interaction_contract": {"agent_channel": {"work_context": context}}}
+    if variant == "incomplete":
+        context["complete"] = False
+    if variant == "unverified":
+        context["selected_todo_text_verified"] = False
+    if variant == "unmatched":
+        content["matched"] = False
+    if variant == "ambiguous":
+        content["ambiguous"] = True
+    if variant == "partial":
+        content["read_complete"] = False
+    if variant == "failed":
+        content["ok"] = False
+    if variant == "wrong-id":
+        todo["todo_id"] = "todo_other"
+    if variant == "wrong-selection":
+        payload["selected_todo"]["todo_id"] = "todo_other"
+    if variant == "duplicate":
+        context["sources"].append(deepcopy(context["sources"][-1]))
+    if variant == "existing-ref":
+        lane["title_ref"] = "caller-owned-reference"
+    if variant == "archived":
+        todo["archive_state"] = "archive"
+    if variant == "done":
+        todo["status"] = "done"
+    if variant == "boolean-id":
+        lane["todo_id"] = todo["todo_id"] = payload["selected_todo"]["todo_id"] = True
+    original = deepcopy(payload)
+    compact = compact_quota_should_run_cli_payload(payload)
+    assert payload == original
+    result = compact["agent_lane_next_action"]
+    if variant in {"text", "title"}:
+        field = "title" if variant == "title" else "text"
+        assert result == {key: value for key, value in lane.items() if key != "title"} | {
+            "title_ref": f"$.interaction_contract.agent_channel.work_context.sources[1].content.todo.{field}"}
+        assert compact["interaction_contract"]["agent_channel"]["work_context"] == context
+        assert todo[field] == title
+        assert len(json.dumps(compact, ensure_ascii=False)) < len(json.dumps(original, ensure_ascii=False)) - len(title) // 2
+        detailed = compact_quota_should_run_cli_payload(payload, include_todo_summary_detail=True)
+        assert detailed["agent_lane_next_action"] == lane
+    else:
+        assert result == lane
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_real_guard_keeps_multiline_task_once_without_differential_opt_in(tmp_path, monkeypatch, provider):
+    from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime, read_fixture_todos
+    from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
+    from test_quota_settlement_cli import AGENT_ID, GOAL_ID, TODO_ID, _run_cli, _write_fixture
+
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    project, runtime, registry = _write_fixture(tmp_path)
+    state = project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md"
+    body = "Keep every requirement and provenance.\n" * 240 + "Stop before an unauthorized publication."
+    todos = read_fixture_todos(state)
+    task = next(todo for todo in todos if todo["todo_id"] == TODO_ID)
+    task.update(text=body, title=body)
+    initialize_canonical_authority(runtime, GOAL_ID,
+        build_todo_runtime_shadow_projection(goal_id=GOAL_ID, todos=todos, handoff_mode="soft_claim"),
+        state_path=state, provider=provider)
+    original_state = state.read_bytes()
+    code, full = _run_cli(registry, runtime, "quota", "should-run", "--goal-id", GOAL_ID,
+        "--agent-id", AGENT_ID, cwd=project)
+    assert code == 0 and full["should_run"], full
+    context = full["interaction_contract"]["agent_channel"]["work_context"]
+    assert context["complete"] and context["selected_todo_text_verified"]
+    index, source = next((i, source) for i, source in enumerate(context["sources"])
+        if source["source"] == "selected_todo")
+    assert source["content"]["todo"]["text"] == body
+    lane = full["agent_lane_next_action"]
+    assert lane["todo_id"] == TODO_ID and "title" not in lane
+    assert lane["title_ref"] == f"$.interaction_contract.agent_channel.work_context.sources[{index}].content.todo.text"
+    assert json.dumps(full, ensure_ascii=False).count(json.dumps(body, ensure_ascii=False)) == 1
+    assert "context_delivery" not in full
+    assert not list(project.rglob("presentation.json"))
+    assert state.read_bytes() == original_state
+    code, detail = _run_cli(registry, runtime, "todo", "list", "--goal-id", GOAL_ID,
+        "--todo-id", TODO_ID, cwd=project)
+    assert code == 0 and detail["todo"]["text"] == body and detail["todo"]["title"] == body
+
+
 @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": [