fix(quota): share complete acceptance within each CLI response

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

@@ -244,6 +244,30 @@ Ordinary product heartbeat consumes current task/acceptance/User sources from
 carrier; the CLI channel carries transitions. TurnEnvelope consumes and signs
 both as `work_context` and `required_reads`; it never independently synthesizes
 another policy. Already delivered content requires no second CLI invocation.
+
+The ordinary `quota should-run` display shares exact complete acceptance copies
+within that response. Its preferred carrier is a loaded Goal-acceptance source
+in `work_context`, otherwise the planning horizon (or the first available full
+copy). Duplicate `acceptance_contract` / `goal_acceptance_contract` fields become
+their corresponding `_ref` fields pointing to that inline carrier. Resolve them
+in the **same response**, including when reading replan context; do not fetch a
+new revision. Different revisions, scopes, sources, failures or JSON types stay
+inline. Nothing is truncated or removed from the retained contract, and no
+decision or acceptance verdict changes. Diagnostic detail options still share
+these copies; canonical state and saved full-decision captures retain their
+original structure. TurnEnvelope relocates the known work-context reference to
+its own signed `work_context` root and rejects an unavailable or dangling source.
+This reduces display bytes, not upstream authority reads, and does not by itself
+establish model adoption or end-to-end efficiency.
+
+中文：普通 `quota should-run` 在同一次响应内复用完全相同的完整验收内容，优先保留
+work context 中已加载的来源，其次保留 planning horizon 或第一份完整副本。重复字段
+改为对应的 `_ref`；读取重规划上下文时也必须沿同响应引用读取，不能另查新版本。
+版本、作用域、来源、失败或 JSON 类型不同则保留原文，不截短要求或改变验收判断。
+详情选项同样复用；权威状态和完整决策捕获不改。TurnEnvelope 将已知引用移到自己
+签名覆盖的 work context，拒绝缺失或不可用来源。这只证明展示成本下降，尚不证明
+上游读取减少、模型采用或完整任务效率提升。
+
 The selected-Todo display body is display-only and may be bounded; the typed
 owner validates the exact read against selected identity, lifecycle, claim and
 source revision. A supplied full-body snapshot must match before fulfilling the read. If the full body is already

loopx/control_plane/quota/cli_projection.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import json
 from typing import Any
 
 from .decision_summary import compact_quota_decision
@@ -704,6 +705,70 @@ def _promote_interaction_contract(payload: dict[str, Any]) -> dict[str, Any]:
     return promoted
 
 
+def _reference_inline_goal_acceptance(payload: dict[str, Any]) -> dict[str, Any]:
+    """Share exact JSON copies in this display only; never trim the contract.
+
+    Work context is the preferred full carrier, then the planning horizon.
+    This Python CLI adapter only changes serialization after owner decisions;
+    canonical state, captures and replan context construction remain untouched.
+    """
+    paths: list[tuple[str | int, ...]] = []
+    interaction = payload.get("interaction_contract")
+    channel = interaction.get("agent_channel") if isinstance(interaction, dict) else None
+    context = channel.get("work_context") if isinstance(channel, dict) else None
+    sources = context.get("sources") if isinstance(context, dict) else None
+    for index, source in enumerate(sources if isinstance(sources, list) else []):
+        if (isinstance(source, dict) and source.get("source") == "goal_acceptance"
+                and isinstance(source.get("content"), dict)
+                and source["content"].get("status", "loaded") == "loaded"
+                and isinstance(source["content"].get("goal_acceptance_contract"), dict)
+                and source["content"]["goal_acceptance_contract"]):
+            paths.append(("interaction_contract", "agent_channel", "work_context",
+                          "sources", index, "content", "goal_acceptance_contract"))
+            break
+    paths.extend([
+        ("planning_horizon", "core_goal", "acceptance_contract"),
+        ("autonomous_replan_obligation", "replan_context", "core_goal", "acceptance_contract"),
+        ("agent_todo_summary", "goal_acceptance_contract"),
+    ])
+    carriers: dict[str, str] = {}
+    result = payload
+    for path in paths:
+        parent: Any = result
+        for part in path[:-1]:
+            if isinstance(parent, dict) and isinstance(part, str):
+                parent = parent.get(part)
+            elif isinstance(parent, list) and isinstance(part, int) and part < len(parent):
+                parent = parent[part]
+            else:
+                parent = None
+                break
+        field = str(path[-1])
+        value = parent.get(field) if isinstance(parent, dict) else None
+        if not isinstance(value, dict) or not value or field + "_ref" in parent:
+            continue
+        # Python equality conflates True/1 and False/0. Compare JSON types and
+        # the complete content, including revision, scope, failures and sources.
+        encoded = json.dumps(value, sort_keys=True, ensure_ascii=False, allow_nan=False)
+        reference = carriers.get(encoded)
+        if reference is None:
+            carriers[encoded] = "$" + "".join(
+                f"[{part}]" if isinstance(part, int) else f".{part}" for part in path)
+            continue
+        if len(encoded) <= len(reference) + 8:
+            continue
+        # Copy only the affected spine, so callers retain their original facts.
+        result = dict(result)
+        target: Any = result
+        for part in path[:-1]:
+            child = target[part]
+            target[part] = list(child) if isinstance(child, list) else dict(child)
+            target = target[part]
+        del target[field]
+        target[field + "_ref"] = reference
+    return result
+
+
 def compact_quota_should_run_cli_payload(
     payload: dict[str, Any],
     *,
@@ -798,6 +863,7 @@ def compact_quota_should_run_cli_payload(
         agent_lane_next_action = dict(agent_lane_next_action)
         agent_lane_next_action.pop("content_revision", None)
         compact["agent_lane_next_action"] = agent_lane_next_action
+    compact = _reference_inline_goal_acceptance(compact)
     return _promote_interaction_contract(
         _promote_runtime_capability_reentry(compact)
     )

loopx/control_plane/quota/turn_envelope.ts

@@ -833,10 +833,31 @@ function turnActionProjection(payload: JsonObject, protocolActionFields: JsonObj
   }
   const acceptance = object(core.acceptance_contract);
   const sources = object(projection.work_context).sources;
+  // The ordinary CLI can already reference its identical work-context copy.
+  // Relocate only this known carrier into the signed envelope. Never carry a
+  // full-decision JSON path into an envelope where that root does not exist.
+  if (!("acceptance_contract" in core) && "acceptance_contract_ref" in core) {
+    const match = typeof core.acceptance_contract_ref === "string"
+      ? /^\$\.interaction_contract\.agent_channel\.work_context\.sources\[(0|[1-9]\d*)\]\.content\.goal_acceptance_contract$/.exec(core.acceptance_contract_ref)
+      : null;
+    const index = match ? Number(match[1]) : -1;
+    const source = object(Array.isArray(sources) ? sources[index] : null);
+    const content = object(source.content);
+    if (!match || source.source !== "goal_acceptance"
+      || (content.status !== undefined && content.status !== "loaded")
+      || Object.keys(object(content.goal_acceptance_contract)).length === 0) {
+      throw new EffectRuntimeRequestError("Goal acceptance reference must resolve to complete inline work context");
+    }
+    action.planning_horizon = {...object(action.planning_horizon), core_goal: {
+      ...core, acceptance_contract_ref: `$.work_context.sources[${index}].content.goal_acceptance_contract`,
+    }};
+    projection.action = action;
+  }
   if (Object.keys(acceptance).length > 0 && Array.isArray(sources)) {
     const index = sources.findIndex((value) => {
       const source = object(value);
       return source.source === "goal_acceptance"
+        && (object(source.content).status === undefined || object(source.content).status === "loaded")
         && canonicalHash(object(object(source.content).goal_acceptance_contract)) === canonicalHash(acceptance);
     });
     if (index >= 0) {

tests/control_plane/test_quota_acceptance_references.py

@@ -0,0 +1,162 @@
+"""Display sharing preserves the complete acceptance and executable decisions."""
+from copy import deepcopy
+import json
+
+import pytest
+
+from loopx.control_plane.quota.cli_projection import (
+    _reference_inline_goal_acceptance, compact_quota_should_run_cli_payload,
+)
+from loopx.control_plane.quota.turn_envelope import (
+    build_turn_envelope, quota_action_signature_document,
+    turn_envelope_action_signature_document,
+)
+from test_replan_host_context_projection import _quota_payload
+
+
+def acceptance():
+    return {"revision": 3, "scope": {"kind": "all_advancement"}, "status": "failed",
+        "requirements": [{"id": "original", "text": "Keep the complete original requirement. " * 20},
+                         {"id": "peer", "text": "Preserve another agent's requirement and stop."}],
+        "sources": [{"ref": "message:original", "summary": "Original owner instruction"}],
+        "verification": {"passed": False, "failures": ["Recovery is still unproven."]}}
+
+
+def packet(with_context=True):
+    source = _quota_payload()
+    contract = acceptance()
+    source["planning_horizon"] = {"core_goal": {"objective": "Deliver the whole outcome",
+                                               "acceptance_contract": deepcopy(contract)}}
+    source["autonomous_replan_obligation"]["replan_context"]["core_goal"] = {
+        "objective": "Deliver the whole outcome", "acceptance_contract": deepcopy(contract)}
+    source["agent_todo_summary"]["goal_acceptance_contract"] = deepcopy(contract)
+    if with_context:
+        source["interaction_contract"]["agent_channel"]["work_context"] = {"complete": True,
+            "sources": [{"source": "goal_acceptance", "content": {"status": "loaded",
+                "provider_revision": "synthetic:3", "goal_acceptance_contract": deepcopy(contract)}}]}
+    return source
+
+
+@pytest.mark.parametrize("with_context", [False, True])
+def test_exact_copies_share_one_complete_source_without_changing_signed_action(with_context):
+    source = packet(with_context)
+    original = deepcopy(source)
+    shared = _reference_inline_goal_acceptance(source)
+    assert source == original
+    assert _reference_inline_goal_acceptance(shared) == shared
+    ref = ("$.interaction_contract.agent_channel.work_context.sources[0].content.goal_acceptance_contract"
+           if with_context else "$.planning_horizon.core_goal.acceptance_contract")
+    assert shared["agent_todo_summary"]["goal_acceptance_contract_ref"] == ref
+    core = shared["autonomous_replan_obligation"]["replan_context"]["core_goal"]
+    assert core == {"objective": "Deliver the whole outcome", "acceptance_contract_ref": ref}
+    if with_context:
+        assert shared["planning_horizon"]["core_goal"]["acceptance_contract_ref"] == ref
+        retained = shared["interaction_contract"]["agent_channel"]["work_context"]["sources"][0]["content"]["goal_acceptance_contract"]
+    else:
+        retained = shared["planning_horizon"]["core_goal"]["acceptance_contract"]
+    assert retained == acceptance()
+    assert len(json.dumps(shared)) < len(json.dumps(source)) - len(json.dumps(retained))
+    assert quota_action_signature_document(shared) == quota_action_signature_document(source)
+    assert turn_envelope_action_signature_document(build_turn_envelope(shared)) == quota_action_signature_document(source)
+    # Default CLI really invokes sharing, including with diagnostic detail flags.
+    for options in ({}, {"include_todo_summary_detail": True, "include_vision_detail": True}):
+        assert compact_quota_should_run_cli_payload(source, **options)["agent_todo_summary"]["goal_acceptance_contract_ref"] == ref
+
+
+@pytest.mark.parametrize("change", ["revision", "scope", "failure", "type", "source"])
+def test_different_contracts_are_retained_in_full(change):
+    source = packet(False)
+    changed = source["autonomous_replan_obligation"]["replan_context"]["core_goal"]["acceptance_contract"]
+    if change == "revision": changed["revision"] = 4
+    if change == "scope": changed["scope"] = {"kind": "selected_work"}
+    if change == "failure": changed["verification"]["failures"] += ["New failure"]
+    if change == "type": changed["verification"]["passed"] = 0  # False == 0 in Python
+    if change == "source": changed["sources"][0]["ref"] = "message:corrected"
+    shared = _reference_inline_goal_acceptance(source)
+    assert shared["autonomous_replan_obligation"]["replan_context"]["core_goal"]["acceptance_contract"] == changed
+    assert "acceptance_contract_ref" not in shared["autonomous_replan_obligation"]["replan_context"]["core_goal"]
+
+
+@pytest.mark.parametrize("missing", [None, {}, [], "unavailable"])
+def test_missing_or_malformed_carriers_never_create_dangling_references(missing):
+    source = packet()
+    source["interaction_contract"]["agent_channel"]["work_context"] = missing
+    source["planning_horizon"] = missing
+    source["autonomous_replan_obligation"] = missing
+    shared = _reference_inline_goal_acceptance(source)
+    assert shared["agent_todo_summary"]["goal_acceptance_contract"] == acceptance()
+    assert "goal_acceptance_contract_ref" not in shared["agent_todo_summary"]
+
+
+def test_unavailable_context_is_not_used_even_if_it_retains_old_content():
+    source = packet()
+    content = source["interaction_contract"]["agent_channel"]["work_context"]["sources"][0]["content"]
+    content["status"] = "unavailable"
+    shared = _reference_inline_goal_acceptance(source)
+    assert shared["planning_horizon"]["core_goal"]["acceptance_contract"] == acceptance()
+    assert shared["agent_todo_summary"]["goal_acceptance_contract_ref"] == "$.planning_horizon.core_goal.acceptance_contract"
+    assert build_turn_envelope(shared)["action"]["planning_horizon"]["core_goal"]["acceptance_contract"] == acceptance()
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_real_cli_reuses_complete_acceptance_for_selection_and_replan(tmp_path, monkeypatch, provider):
+    from test_replan_successor_guard_reentry import _fixture, AGENT, GOAL
+    monkeypatch.setenv("LOOPX_USAGE_PING", "0")
+    call, runtime, index = _fixture(tmp_path, monkeypatch, provider, future_monitor=False)
+    registry_path = tmp_path / "registry.json"
+    registry = json.loads(registry_path.read_text())
+    registry["goals"][0]["execution_profile"] = {"planning_horizon_goal": True}
+    registry_path.write_text(json.dumps(registry))
+    document = tmp_path / "requirements.json"
+    document.write_text(json.dumps({"outcome_id": "export", "owner_agent_id": AGENT,
+        "objective": "Deliver the complete export", "requirements": [
+            {"id": "content", "description": "Preserve every original section."},
+            {"id": "recovery", "description": "Keep the tail requirement and the publication stop."}],
+        "source": {"ref": "message:owner", "summary": "The owner requested a complete export."}}))
+    before = call("goal-acceptance", "inspect", "--goal-id", GOAL)
+    call("goal-acceptance", "refine", "--goal-id", GOAL, "--agent-id", AGENT,
+         "--document", str(document), "--expected-provider-revision", before["provider_revision"],
+         "--operation-id", "source-requirements", "--execute")
+    expected = call("goal-acceptance", "inspect", "--goal-id", GOAL)["goal_acceptance_contract"]
+    guard = call("quota", "should-run", "--codex-app", "--goal-id", GOAL, "--agent-id", AGENT,
+                 "--turn-instance-id", "turn-acceptance-sharing")
+    assert guard["effective_action"] == "autonomous_replan_required"
+    ref = "$.interaction_contract.agent_channel.work_context.sources[0].content.goal_acceptance_contract"
+    assert guard["autonomous_replan_obligation"]["replan_context"]["core_goal"]["acceptance_contract_ref"] == ref
+    assert guard["agent_todo_summary"]["goal_acceptance_contract_ref"] == ref
+    assert guard["interaction_contract"]["agent_channel"]["work_context"]["sources"][0]["content"]["goal_acceptance_contract"] == expected
+    assert build_turn_envelope(guard)["action_signature"]["matches"] is True
+
+    # Normal selected delivery must use the exact inline acceptance read and
+    # relocate its reference when a host consumes the CLI as a TurnEnvelope.
+    index.write_text("")
+    added = call("todo", "add", "--goal-id", GOAL, "--role", "agent", "--claimed-by", AGENT,
+                 "--text", "Validate the complete export", "--task-class", "advancement_task",
+                 "--action-kind", "validate")
+    call("todo", "add", "--goal-id", GOAL, "--role", "agent", "--claimed-by", AGENT,
+         "--text", "Validate independent recovery", "--task-class", "advancement_task",
+         "--action-kind", "validate")
+    linked = json.loads(document.read_text())
+    linked["todo_ids"] = [added["todo_id"]]
+    document.write_text(json.dumps(linked))
+    revision = call("goal-acceptance", "inspect", "--goal-id", GOAL)["provider_revision"]
+    call("goal-acceptance", "refine", "--goal-id", GOAL, "--agent-id", AGENT,
+         "--document", str(document), "--expected-provider-revision", revision,
+         "--operation-id", "link-export", "--execute")
+    expected = call("goal-acceptance", "inspect", "--goal-id", GOAL)["goal_acceptance_contract"]
+    planning = call("quota", "should-run", "--codex-app", "--goal-id", GOAL, "--agent-id", AGENT,
+                    "--turn-instance-id", "turn-export-planning")
+    assert planning["planning_horizon"]["core_goal"]["acceptance_contract"] == expected
+    assert planning["agent_todo_summary"]["goal_acceptance_contract_ref"] == "$.planning_horizon.core_goal.acceptance_contract"
+    selected = call("quota", "should-run", "--codex-app", "--goal-id", GOAL, "--agent-id", AGENT,
+                    "--todo-id", added["todo_id"], "--turn-instance-id", "turn-export-delivery")
+    assert selected["normal_delivery_allowed"] is True
+    sources = selected["interaction_contract"]["agent_channel"]["work_context"]["sources"]
+    source_index = next(i for i, row in enumerate(sources) if row["source"] == "goal_acceptance")
+    retained = sources[source_index]["content"]["goal_acceptance_contract"]
+    assert retained == expected
+    assert selected["agent_todo_summary"]["goal_acceptance_contract_ref"] == (
+        f"$.interaction_contract.agent_channel.work_context.sources[{source_index}].content.goal_acceptance_contract")
+    envelope = build_turn_envelope(selected)
+    assert envelope["work_context"]["sources"][source_index]["content"]["goal_acceptance_contract"] == expected
+    assert envelope["action_signature"]["matches"] is True