fix(quota): retain composed settlement failure diagnostics

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

@@ -269,12 +269,20 @@ When `--delivery-workspace-path` names the producing worktree, the combined
 operation also runs the owed command there, using the originating CLI's
 implementation. The normal workspace and settlement guards still apply; a
 missing or incompatible workspace leaves the original owed command recoverable.
+In JSON, a failed spend's response is available under `quota_spend`, including
+any `workspace_guard` remediation or bounded child reader. For a workspace
+refusal, use the guard's required action and the original `settlement_owed`
+command. Use the parent's verified receipts to determine what committed.
 
 Callers that need a compact result can add `--settlement-receipt` to
 `refresh-state --settle-turn`. A successful settled result provides its exact
 identity and receipts inline and a `result_details.command` for the full saved
 result. Inspect unresolved controls before continuing. If saving the view fails
 after commitment, retry the same Turn and check its receipts.
+If a failed result exceeds the receipt budget, read the complete selected parent
+controls first and follow any required reader under `quota_spend`. If the
+failure capture cannot be saved, the result returns its original failure controls
+with `result_delivery_error`.
 
 An in-flight recommendation does not require keeping a completed Todo open.
 Validate its original declaration, use ordinary completion when satisfied, and
@@ -297,10 +305,15 @@ Todo 或 obligation/Turn 的 canonical 待结算命令并验证回执；不重
 指定 `--delivery-workspace-path` 后，组合操作也在该交付工作区执行待结算命令，
 继续使用原调用方的 CLI 实现。原工作区和结算校验保持；目录缺失或不兼容时保留
 原待结算命令以便恢复。
+JSON 的 `quota_spend` 保留子扣额的失败响应，包括工作区 guard 的恢复动作和已有的
+分页读取入口。工作区拒绝时，按 guard 的要求和原 `settlement_owed` 命令恢复，并以父层
+核验的回执判断已提交的效果。
 
 需要紧凑回执时，在 `refresh-state --settle-turn` 后加 `--settlement-receipt`。
 结算成功后可直接核对本次身份和回执，用 `result_details.command` 读取完整结果。
 先处理未解决的控制项。若结算已提交而视图保存失败，按原 Turn 重试并核对回执。
+失败结果超过回执预算时，先完整读取父层选中的控制项，再按 `quota_spend` 中的要求
+继续读取子结果。若失败结果无法保存，响应保留原控制项并返回 `result_delivery_error`。
 
 执行中续跑建议不要求把已完成的 Todo 保持开放。原声明验收满足后可正常完成，并
 在原 Turn 上按语义收尾写入有证据的 vision；未完成的阶段进展仍可保留开放状态。

loopx/cli_commands/quota_capture.py

@@ -1010,14 +1010,18 @@ def fits(value: dict[str, object]) -> bool:
         # Large or unknown controls are mandatory selected reads, never cut text.
         pointers = ["/" + key.replace("~", "~0").replace("/", "~1")
             for key in payload if key not in deferred]
+        snapshot_note = ""
+        if supporting_fields is not None:
+            snapshot_note = " Supporting snapshots remain available in the same immutable result."
+        elif deferred:
+            snapshot_note = " before/after remain available in the same immutable result."
         view = {"ok": payload.get("ok"), "result_details": details,
             "command": context_capture_read_command(saved, digest, reference_file=reference, select=pointers,
                 page_bytes=min(NATIVE_CONTEXT_PAGE_BYTES, output_bytes),
                 max_page_output_bytes=output_bytes, reader_argv=reader_argv),
             "instruction": "Read the complete selected result controls before acting. The operation was "
                 "already evaluated; preserve its original Turn/effect identity. No execution authority "
-                "or context baseline is granted. " + ("Supporting snapshots remain available in the same immutable result."
-                    if supporting_fields is not None else "before/after remain available in the same immutable result.")}
+                "or context baseline is granted." + snapshot_note}
         if fits(view):
             return view
         raise ValueError(f"result reader metadata exceeds the output budget; complete result: {saved}")

loopx/cli_commands/refresh_settlement.py

@@ -90,6 +90,12 @@ def settle_refreshed_turn(
                 error = str(spent.get("error") or spent.get("reason") or "quota settlement CLI failed")
         except (OSError, RuntimeError, ValueError, TypeError, KeyError, subprocess.TimeoutExpired, SystemExit) as exc:
             error = str(exc)
+    # A refused or partially delivered child can carry recovery controls and
+    # its own bounded reader. Preserve that view before attempting readback.
+    payload["quota_spend"] = spent if error is not None else {key: spent[key] for key in (
+        "ok", "appended", "idempotent_replay", "json_path", "rollout_event",
+        "reward_memory_ingest", "error", "error_code",
+    ) if key in spent}
     try:
         readback = read_heartbeat_settlement(
             runtime_root, registry_path=registry_path, goal_ref=goal_ref,
@@ -107,10 +113,6 @@ def settle_refreshed_turn(
         return
     attach_settlement_progress(payload, readback, registry_path=registry_path,
         runtime_root=runtime_root, goal_ref=goal_ref)
-    payload["quota_spend"] = {key: spent[key] for key in (
-        "ok", "appended", "idempotent_replay", "json_path", "rollout_event",
-        "reward_memory_ingest", "error", "error_code",
-    ) if key in spent}
     payload["settlement_result"] = settlement_result_payload(readback.settlement)
     if payload["settlement_progress"]["state"] == "spend_required":
         payload.update(ok=False, error=error or "exact quota receipt is uncommitted; recover the returned original settlement_owed")
@@ -123,21 +125,34 @@ def settle_refreshed_turn(
 
 
 def present_settled_refresh(payload: dict[str, Any], *, runtime_root: Path) -> dict[str, Any]:
-    """Keep controls/receipts inline; retain successful diagnostic snapshots.
+    """Keep the receipt view bounded and retain complete recovery controls.
 
-    This opt-in output adapter runs after the original owners and exact receipt
-    readback. Errors, pending recovery and unknown fields remain fully visible.
-    It neither changes a result nor certifies model-context retention.
+    This opt-in output adapter follows the original owners and receipt readback.
+    A wide result requires reading the complete controls before acting.
     """
+    if payload.get("dry_run"):
+        return payload
     progress = payload.get("settlement_progress")
-    if (payload.get("ok") is not True or payload.get("dry_run")
+    failed_with_spend = payload.get("ok") is False and bool(payload.get("quota_spend"))
+    if not failed_with_spend and (payload.get("ok") is not True
             or not isinstance(progress, dict) or progress.get("state") != "settled"
             or payload.get("settlement_owed") is not None):
         return payload
     from .quota_capture import budgeted_settlement_result
     from ..command_invocation import current_loopx_cli_argv
     from ..control_plane.turn_driver.native_context_profile import NATIVE_CONTEXT_PAGE_OUTPUT_BYTES
 
+    reader_argv = current_loopx_cli_argv()
+    if failed_with_spend:
+        from ..control_plane.quota.error_codes import QuotaDecisionTransportError
+
+        try:
+            return budgeted_settlement_result(payload, runtime_root=runtime_root,
+                output_bytes=NATIVE_CONTEXT_PAGE_OUTPUT_BYTES, reader_argv=reader_argv)
+        except QuotaDecisionTransportError as exc:
+            # Retain the original refusal and retry controls if capture fails.
+            return {**payload, "result_delivery_error": {"error_code": exc.error_code, "reason": str(exc)}}
+
     supporting = ["registry", "runtime_root", "project", "json_path", "markdown_path", "index_path",
                   "health_check", "state", "registry_goal", "todo_source"]
     # Only fully delivered projection diagnostics are supporting material.
@@ -216,4 +231,4 @@ def present_settled_refresh(payload: dict[str, Any], *, runtime_root: Path) -> d
             supporting.append(field)
     return budgeted_settlement_result(payload, runtime_root=runtime_root,
         output_bytes=NATIVE_CONTEXT_PAGE_OUTPUT_BYTES, supporting_fields=tuple(supporting),
-        reader_argv=current_loopx_cli_argv())
+        reader_argv=reader_argv)

tests/control_plane/test_refresh_settle_turn.py

@@ -5,6 +5,7 @@
 import copy
 import shlex
 import subprocess
+import sys
 
 import pytest
 
@@ -44,6 +45,17 @@ def test_receipt_capture_failure_keeps_committed_receipts_and_exact_retry(tmp_pa
     assert _spend_run_count(runtime) == 1
 
 
+def _read_selected_result(command):
+    bodies = []
+    while command:
+        result = subprocess.run(shlex.split(command), capture_output=True, text=True, check=False)
+        page = json.loads(result.stdout)
+        assert result.returncode == 0 and len(result.stdout.encode()) <= 64000, page
+        bodies.append(page["content"])
+        command = page.get("next_command")
+    return {row["pointer"]: row["value"] for row in json.loads("".join(bodies))["selections"]}
+
+
 def test_receipt_keeps_unknown_controls_and_never_hides_partial_failure(tmp_path):
     from loopx.cli_commands.refresh_settlement import present_settled_refresh
 
@@ -62,17 +74,36 @@ def test_receipt_keeps_unknown_controls_and_never_hides_partial_failure(tmp_path
     bounded = present_settled_refresh(large, runtime_root=tmp_path)
     assert "Read the complete selected result controls before acting" in bounded["instruction"]
     assert "Supporting snapshots" in bounded["instruction"]
-    command, bodies = bounded["command"], []
-    while command:
-        result = subprocess.run(shlex.split(command), capture_output=True, text=True, check=False)
-        page = json.loads(result.stdout)
-        assert result.returncode == 0 and len(result.stdout.encode()) <= 64000
-        bodies.append(page["content"])
-        command = page.get("next_command")
-    selected = {row["pointer"]: row["value"] for row in json.loads("".join(bodies))["selections"]}
+    selected = _read_selected_result(bounded["command"])
     assert selected["/future_control"] == large["future_control"]
 
 
+def test_wide_failed_spend_keeps_nested_reader_and_parent_controls(tmp_path):
+    from loopx.cli_commands.quota_capture import budgeted_settlement_result
+    from loopx.cli_commands.refresh_settlement import present_settled_refresh
+
+    child = {"ok": False, "appended": False, "future_recovery_control": {"detail": "step " * 15000}}
+    child_view = budgeted_settlement_result(child, runtime_root=tmp_path, output_bytes=64000,
+        reader_argv=[sys.executable, "-m", "loopx.entrypoint"])
+    parent = {"ok": False, "error": "Original refusal", "quota_spend": child_view,
+        "settlement_identity": {"effect_id": "original-effect"},
+        "settlement_owed": {"command": "original owed command"},
+        "another_control": "detail " * 10000}
+    bounded = present_settled_refresh(parent, runtime_root=tmp_path)
+    assert bounded["ok"] is False and "Read the complete selected result controls" in bounded["instruction"]
+    selected = _read_selected_result(bounded["command"])
+    assert selected["/settlement_identity"] == parent["settlement_identity"]
+    assert selected["/settlement_owed"] == parent["settlement_owed"]
+    assert selected["/quota_spend"] == child_view
+    child_controls = _read_selected_result(selected["/quota_spend"]["command"])
+    assert child_controls["/future_recovery_control"] == child["future_recovery_control"]
+
+    (tmp_path / "context-delivery").chmod(0o755)
+    fallback = present_settled_refresh(parent, runtime_root=tmp_path)
+    assert fallback.pop("result_delivery_error")["error_code"] == "quota_decision_transport_failed"
+    assert fallback == parent
+
+
 def _repository_delivery(tmp_path, provider):
     project, runtime, registry = _write_fixture(tmp_path)
     state = project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md"
@@ -124,8 +155,6 @@ def test_combined_settlement_uses_producing_worktree_and_originating_cli(
 def test_wrong_or_missing_child_workspace_retains_owed_command_for_exact_recovery(
     tmp_path, monkeypatch, provider,
 ):
-    from loopx.cli_commands.refresh_settlement import settle_refreshed_turn
-
     isolate_sqlite_runtime(tmp_path, monkeypatch)
     project, runtime, registry, delivery = _repository_delivery(tmp_path, provider)
     binding = ["--goal-id", GOAL_ID, "--agent-id", AGENT_ID,
@@ -134,21 +163,28 @@ def test_wrong_or_missing_child_workspace_retains_owed_command_for_exact_recover
     assert rc == 0, guard
     refresh = ["refresh-state", *binding, "--classification", "verified_worktree",
                "--delivery-batch-scale", "implementation", "--delivery-outcome", "outcome_progress",
-               "--delivery-boundary", "in_flight_continuation", "--no-global-sync", "--suppress-external-sinks",
-               "--delivery-workspace-path", str(delivery)]
-    rc, original = _run_cli(registry, runtime, *refresh, cwd=project)
+               "--delivery-boundary", "in_flight_continuation", "--no-global-sync", "--suppress-external-sinks"]
+    producing = [*refresh, "--delivery-workspace-path", str(delivery)]
+    rc, original = _run_cli(registry, runtime, *producing, cwd=project)
     assert rc == 0 and _spend_run_count(runtime) == 0, original
     for workspace in (project, tmp_path / "missing"):
-        failed = copy.deepcopy(original)
-        settle_refreshed_turn(failed, registry_path=registry, runtime_root=runtime,
-            goal_id=GOAL_ID, agent_id=AGENT_ID, todo_id=TODO_ID,
-            replan_obligation_id=None, turn_instance_id=TURN_ID, goal_ref=None,
-            delivery_workspace_path=workspace)
-        assert failed["ok"] is False, failed
+        rc, failed = _run_cli(registry, runtime, *refresh, "--delivery-workspace-path", str(workspace),
+            "--settle-turn", "--settlement-receipt", cwd=project)
+        assert rc != 0 and failed["ok"] is False, failed
         assert failed["settlement_progress"]["state"] == "spend_required"
+        assert failed["settlement_identity"] == original["settlement_identity"]
         assert failed["settlement_owed"]["command"] == original["settlement_owed"]["command"]
         assert _spend_run_count(runtime) == 0
-    rc, recovered = _run_cli(registry, runtime, *refresh, "--settle-turn", cwd=project)
+        if workspace == project:
+            child = failed["quota_spend"]
+            assert child["ok"] is False and child["appended"] is False
+            workspace_guard = child["workspace_guard"]
+            assert workspace_guard["action"] == "return_to_delivery_worktree"
+            assert workspace_guard["current_workspace"] == "canonical_checkout"
+            assert workspace_guard["blocks_delivery"] is True
+            assert "workspace that produced" in workspace_guard["required_action"]
+            assert child["before"]["selected_todo"]["todo_id"] == TODO_ID and child["after"] is None
+    rc, recovered = _run_cli(registry, runtime, *producing, "--settle-turn", cwd=project)
     assert rc == 0, recovered
     assert recovered["settlement_progress"]["state"] == "settled"
     assert recovered["settlement_identity"] == original["settlement_identity"]
@@ -331,6 +367,7 @@ def lose_response(command, **kwargs):
     result = copy.deepcopy(original)
     settle(result)
     assert result["ok"] and result["settlement_progress"]["state"] == "settled", result
+    assert "Synthetic response loss" in result["quota_spend"]["transport_warning"]
     assert len(calls) == 1 and _spend_run_count(runtime) == 1
     monkeypatch.setattr(refresh_settlement.subprocess, "run", real_run)
     replay = copy.deepcopy(original)