Reduce repeated Explore input in refresh-state JSON results

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

@@ -31,6 +31,26 @@ TypeScript 上下文 owner 同时约束 `quota context-read --initial-read`。
 
 ## Ownership Boundary
 
+Successful JSON `refresh-state` output keeps the complete `explore_result` once.
+When a successful Explore hook's ingestion intent contains exactly the same
+attachment, its display payload uses `attachment_ref: "$.explore_result"` within
+that response. Other intent fields, delivery results, settlement identity and
+receipts remain intact. Numeric spellings participate in equality, so adjacent
+large integers cannot be mistaken for identical inputs. Failed/pending delivery,
+conflicts, unknown schemas and existing references stay inline. If optional
+display transport fails, the complete original response is printed. Markdown,
+persisted intents and exact replay inputs are unchanged; no capture or additional
+model read is required. This reduces repeated output, not semantic/backend work,
+and does not establish lower fees or better model outcomes.
+
+成功的 JSON `refresh-state` 输出完整保留一份 `explore_result`。Explore hook
+成功交付、且 ingestion intent 的附件与其完全相同时，仅显示副本替换为同一响应内的
+`attachment_ref: "$.explore_result"`；其他输入、交付结果、结算身份和回执保持原样。
+相等判断保留原始数值表示，避免将相邻大整数误判为相同。失败、待重试、内容冲突、
+未知 schema 和已有引用保持内联；可选显示传输失败则打印原完整结果。Markdown、
+持久化 intent 和精确重放输入不变，无需新增 capture 或模型补读。这只减少重复输出，
+不减少语义／后端工作，也不证明费用或模型效果改善。
+
 The quota control plane owns decision, precedence, scheduler, interaction,
 selected-todo, and user-action semantics. `cli_projection.py` owns only the
 serialized view consumed by agents. A compactor must not become a second

loopx/cli_commands/project_lifecycle_refresh_state.py

@@ -80,6 +80,7 @@
     reward_memory_candidate_details,
     stage_reward_memory_outcome_candidate,
 )
+from ..control_plane.quota.cli_projection import compact_refresh_state_cli_payload
 
 
 def register_refresh_state_command(
@@ -871,7 +872,7 @@ def handle_refresh_state_command(
             )
         if not material_refresh_ready:
             _settle_requested_refresh(payload, args, registry_path, goal_ref)
-            print_payload(payload, fmt, render_state_refresh_markdown)
+            print_payload(compact_refresh_state_cli_payload(payload) if fmt == "json" else payload, fmt, render_state_refresh_markdown)
             return 0 if payload.get("ok") else 1
         if payload.get("explore_result") is not None:
             from ..capabilities.explore.result_writeback import deliver_result_attachment
@@ -961,7 +962,7 @@ def handle_refresh_state_command(
             ),
         )
     _settle_requested_refresh(payload, args, registry_path, goal_ref)
-    print_payload(payload, fmt, render_state_refresh_markdown)
+    print_payload(compact_refresh_state_cli_payload(payload) if fmt == "json" else payload, fmt, render_state_refresh_markdown)
     return 0 if payload.get("ok") else 1
 
 

loopx/control_plane/capabilities/explore_result_writeback.ts

@@ -2,7 +2,8 @@
 import {createHash} from "node:crypto";
 import type {JsonObject} from "../effect_program.ts";
 import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts";
-import {requireJsonObject, requireNonEmptyString, requireStringArray, requireStringLiteral} from "../runtime_decode.ts";
+import {jsonObject, requireJsonObject, requireNonEmptyString, requireStringArray, requireStringLiteral} from "../runtime_decode.ts";
+import {contextSourceIdentity} from "../quota/diagnostic_projection.ts";
 import {normalizeGoalPathDelta} from "../goals/vision_checkpoint.ts";
 import {normalizeExploreExecutionObservation} from "./explore_execution_observation.ts";
 import {isExploreEvidenceRef} from "./explore_evidence_ref.ts";
@@ -329,3 +330,32 @@ export function produceExploreResultIntent(params: JsonObject): JsonObject {
       idempotency_key: `explore-result:${key}`, source_receipt_id: source,
       payload: {attachment}, requested_write_scope: []}};
 }
+
+/** Only the final display shares the successful hook's identical input. The
+ * complete attachment stays inline; durable intents and retry inputs never change. */
+export function referenceExploreResultDelivery(source: JsonObject, numberTokens?: unknown): JsonObject {
+  const attachment = jsonObject(source.explore_result), delivery = jsonObject(source.explore_result_delivery);
+  const hook = jsonObject(delivery?.hook);
+  if (attachment?.schema_version !== EXPLORE_RESULT_ATTACHMENT_SCHEMA ||
+      Object.hasOwn(attachment, "attachment_ref") || delivery?.ok !== true ||
+      delivery.retryable !== false || delivery.primary_committed !== true ||
+      hook?.schema_version !== "loopx_post_writeback_capability_hook_dispatch_v0" ||
+      hook.phase !== "post_writeback" || !Array.isArray(hook.failures) || hook.failures.length ||
+      !Array.isArray(hook.intents)) return source;
+  const identity = contextSourceIdentity(source, numberTokens);
+  const original = identity(attachment, ["explore_result"]);
+  const reference = {attachment_ref: "$.explore_result"};
+  if (Buffer.byteLength(JSON.stringify({attachment})) <= Buffer.byteLength(JSON.stringify(reference))) return source;
+  let changed = false;
+  const intents = hook.intents.map((value, index) => {
+    const intent = jsonObject(value), payload = jsonObject(intent?.payload);
+    if (intent?.schema_version !== "loopx_capability_intent_v0" ||
+        intent.intent_kind !== "explore.result_ingestion" || !payload ||
+        Object.hasOwn(payload, "attachment_ref") || !jsonObject(payload.attachment) ||
+        identity(payload.attachment, ["explore_result_delivery", "hook", "intents", index, "payload", "attachment"]) !== original) return value;
+    const {attachment: _duplicate, ...other} = payload;
+    changed = true;
+    return {...intent, payload: {...other, ...reference}};
+  });
+  return changed ? {...source, explore_result_delivery: {...delivery, hook: {...hook, intents}}} : source;
+}

loopx/control_plane/quota/cli_projection.py

@@ -627,6 +627,25 @@ def _apply_context_projection_changes(
     return cast(dict[str, Any], projected)
 
 
+def compact_refresh_state_cli_payload(payload: dict[str, Any]) -> dict[str, Any]:
+    """Share only repeated successful Explore input; full output is the fallback."""
+    if (payload.get("ok") is not True or not isinstance(payload.get("explore_result"), dict)
+            or not isinstance(payload.get("explore_result_delivery"), dict)):
+        return payload
+    from ..effect_runtime import source_number_tokens
+    from .turn_envelope import _turn_envelope_result
+
+    source = {key: payload[key] for key in ("explore_result", "explore_result_delivery")}
+    try:
+        result = _turn_envelope_result("reference_context", payload=source,
+            refresh_result=True, source_number_tokens=source_number_tokens(source))
+        return {**payload, **_apply_context_projection_changes(source, result)}
+    except (RuntimeError, ValueError, OSError):
+        # A display failure must not obscure a committed result or suggest that
+        # the caller should repeat admission, writeback or accounting.
+        return payload
+
+
 def compact_quota_should_run_cli_payload(
     payload: dict[str, Any],
     *,

loopx/control_plane/quota/turn_envelope.ts

@@ -8,6 +8,7 @@ import { ENVELOPED_SHA256_PATTERN } from "../content_digest.ts";
 import {omitUnboundExploreResultWriteback, referenceExploreResultWriteback} from "../capabilities/explore_turn_context.ts";
 import {prepareContextDelivery, restoreContextDelivery} from "./context_delivery.ts";
 import {updateNativeContextLifecycle} from "./native_context_lifecycle.ts";
+import {referenceExploreResultDelivery} from "../capabilities/explore_result_writeback.ts";
 
 import {
   interpretQuotaShouldRunPacket,
@@ -1487,6 +1488,7 @@ export function evaluateTurnEnvelope(value: unknown): JsonObject {
     if (jsonObject(request.diagnostics)) payload = projectQuotaDiagnostics(payload, request.diagnostics as JsonObject);
     if (request.task_title === true) payload = referenceCurrentTaskTitle(payload);
     if (request.explore_guidance === true) payload = referenceExploreGuidance(payload);
+    if (request.refresh_result === true) payload = referenceExploreResultDelivery(payload, request.source_number_tokens);
     // References and diagnostic caps share one edit transport. Untouched facts
     // stay in the original caller, including numbers JSON.parse cannot retain.
     return contextProjectionChanges(source, payload);

tests/control_plane/test_explore_result_writeback.py

@@ -880,6 +880,9 @@ def run(*command):
     current = run("task-lease", "acquire", *lease_binding, "--idempotency-key", "recovery")["lease"]
     captured = run(*refresh)
     assert captured["appended"] and captured["explore_result_delivery"]["ok"]
+    display_input = captured["explore_result_delivery"]["hook"]["intents"][0]["payload"]
+    assert display_input == {"attachment_ref": "$.explore_result"}
+    assert captured["explore_result"] == ATTACHMENT
     assert len(load_explore_result_events(log)) == 2
     committed_index, committed_log = index.read_bytes(), log.read_bytes()
     rows = list_goal_todos(registry_path=path, goal_id=GOAL_ID)["todos"]
@@ -893,6 +896,8 @@ def run(*command):
     assert leases[0]["status"] == "released"
     replay = run(*refresh)
     assert replay["idempotent_replay"] and replay["explore_result_delivery"]["idempotent_replay"]
+    assert replay["explore_result"] == captured["explore_result"]
+    assert replay["explore_result_delivery"]["hook"]["intents"][0]["payload"] == display_input
     assert index.read_bytes() == committed_index and log.read_bytes() == committed_log
     assert list_goal_todos(registry_path=path, goal_id=GOAL_ID)["todos"] == rows
     assert read_canonical_todos_if_promoted(

tests/control_plane/test_refresh_result_projection.py

@@ -0,0 +1,90 @@
+"""Successful refresh display shares input without changing any result fact."""
+from copy import deepcopy
+import json
+
+import pytest
+
+from loopx.control_plane.quota.cli_projection import compact_refresh_state_cli_payload
+from test_explore_result_writeback import ATTACHMENT
+
+
+def packet():
+    attachment = {**ATTACHMENT, "future_measurement": {"value": 2**60 + 1, "stop": "Keep the original stop."}}
+    return {"ok": True, "settlement_identity": {"effect_id": "original-turn"},
+        "settlement_progress": {"state": "settled"}, "future_control": {"stop": True},
+        "explore_result": attachment, "explore_result_delivery": {
+            "ok": True, "retryable": False, "primary_committed": True, "future_delivery": [0, False],
+            "hook": {"schema_version": "loopx_post_writeback_capability_hook_dispatch_v0",
+                "phase": "post_writeback", "failures": [], "intent_count": 1, "intents": [{
+                    "schema_version": "loopx_capability_intent_v0", "intent_kind": "explore.result_ingestion",
+                    "idempotency_key": "original-intent", "source_receipt_id": "original-receipt",
+                    "payload": {"attachment": deepcopy(attachment), "future_input": {"retain": True}},
+                    "requested_write_scope": []}]}}}
+
+
+def intent_payload(value):
+    return value["explore_result_delivery"]["hook"]["intents"][0]["payload"]
+
+
+@pytest.mark.parametrize("value", [2**60 + 1, 1.0, -0.0])
+def test_all_result_facts_are_reconstructable_in_one_response_without_mutation(value):
+    source = packet()
+    source["explore_result"]["future_measurement"]["value"] = value
+    intent_payload(source)["attachment"]["future_measurement"]["value"] = value
+    original = json.dumps(source)
+    shared = compact_refresh_state_cli_payload(source)
+    assert intent_payload(shared)["attachment_ref"] == "$.explore_result"
+    restored = deepcopy(shared)
+    target = intent_payload(restored)
+    assert target.pop("attachment_ref") == "$.explore_result"
+    target["attachment"] = restored["explore_result"]
+    assert restored == source
+    assert json.dumps(source) == original
+    assert len(json.dumps(shared).encode()) < len(original.encode())
+    assert compact_refresh_state_cli_payload(shared) == shared
+
+
+@pytest.mark.parametrize("mutation", ["failed", "pending", "hook-failed", "conflict", "numeric-collision",
+    "unknown-hook", "unknown-intent", "unknown-attachment", "preexisting-ref", "missing-source"])
+def test_failures_conflicts_unknown_schemas_and_reference_collisions_stay_complete(mutation):
+    source = packet()
+    delivery = source["explore_result_delivery"]
+    if mutation == "failed":
+        source["ok"] = False
+    elif mutation == "pending":
+        delivery.update(ok=False, retryable=True, error="Replay the original writeback.")
+    elif mutation == "hook-failed":
+        delivery["hook"]["failures"] = [{"error_code": "future_failure"}]
+    elif mutation == "conflict":
+        intent_payload(source)["attachment"]["interpretation"] += " Additional stop."
+    elif mutation == "numeric-collision":
+        intent_payload(source)["attachment"]["future_measurement"]["value"] = 2**60
+    elif mutation == "unknown-hook":
+        delivery["hook"]["schema_version"] = "future_hook_v1"
+    elif mutation == "unknown-intent":
+        delivery["hook"]["intents"][0]["intent_kind"] = "future_ingestion"
+    elif mutation == "unknown-attachment":
+        source["explore_result"]["schema_version"] = "future_attachment_v1"
+    elif mutation == "preexisting-ref":
+        intent_payload(source)["attachment_ref"] = "$.other_source"
+    else:
+        del source["explore_result"]
+    original = json.dumps(source)
+    assert json.dumps(compact_refresh_state_cli_payload(source)) == original
+
+
+def test_disabled_path_needs_no_projection_and_display_failure_retains_committed_receipt(monkeypatch):
+    from loopx.control_plane.quota import turn_envelope
+    calls = []
+
+    def unavailable(*args, **kwargs):
+        calls.append(args)
+        raise RuntimeError("display transport unavailable")
+
+    monkeypatch.setattr(turn_envelope, "_turn_envelope_result", unavailable)
+    off = {"ok": True, "settlement_progress": {"state": "settled"}}
+    assert compact_refresh_state_cli_payload(off) is off
+    assert calls == []
+    source = packet()
+    assert compact_refresh_state_cli_payload(source) is source
+    assert len(calls) == 1

tests/control_plane/test_refresh_settle_turn.py

@@ -112,7 +112,7 @@ def test_same_turn_progress_or_completion_and_replay_without_extra_guard(
     plan = guard["interaction_contract"]["cli_channel"]["settlement_plan"]
     validation, writeback = plan["ordered_steps"][:2]
     assert validation["command_condition"] == "todo_deliverable_complete"
-    assert "semantic_closeout" in validation["precondition"]
+    assert " --delivery-boundary semantic_closeout " in writeback["semantic_closeout_command_template"]
     vision_args = []
     if finish_todo:
         rc, complete = _run_generated_cli(validation["command_template"].replace(