Keep durable quota events out of oversized settlement errors

loopx/cli_commands/quota.py

@@ -20,6 +20,7 @@
     bind_capture_selection_transport, budgeted_settlement_result, capture_decision, handoff_decision,
     handle_native_context_command, prepare_quota_capture,
     native_context_activation, progressive_context_or_current, project_native_quota_context,
+    settlement_capture_failure_view,
 )
 from ..control_plane.turn_driver.decision_capture import (
     project_captured_context,
@@ -317,15 +318,7 @@ def _emit_bounded_settlement(payload, args, *, runtime_root, reader_argv, print_
         view = budgeted_settlement_result(payload, runtime_root=runtime_root,
             output_bytes=args.max_page_output_bytes, reader_argv=reader_argv)
     except QuotaDecisionTransportError as exc:
-        # Transport failure cannot replace an already evaluated refusal or
-        # committed receipt. Preserve unknown controls as well as identity;
-        # only the bulky diagnostic snapshots are omitted on this fallback.
-        view = {key: value for key, value in payload.items() if key not in ("before", "after")}
-        view["result_delivery_error"] = {"error_code": exc.error_code, "reason": str(exc),
-            "recommended_action": "Preserve the original result controls and Turn identity. "
-                "Repair capture privacy or remove --max-page-output-bytes and read back the "
-                "same operation; do not repeat admission or assume it did not commit.",
-            "omitted_diagnostics": [key for key in ("before", "after") if key in payload]}
+        view = settlement_capture_failure_view(payload, output_bytes=args.max_page_output_bytes, error=exc)
         print_payload(view, args.format, _quota_renderer(args, failure=True))
         return 1
     print_payload(view, args.format, _quota_renderer(args, failure=payload.get("ok") is not True))

loopx/cli_commands/quota_capture.py

@@ -649,6 +649,29 @@ def capture_decision(directory: Path | None, payload: dict[str, object], *,
         ) from exc
 
 
+def settlement_capture_failure_view(
+    payload: dict[str, object], *, output_bytes: int, error: QuotaDecisionTransportError,
+) -> dict[str, object]:
+    """Keep the evaluated result controls available when capture fails."""
+    omitted = [key for key in ("before", "after") if key in payload]
+    view = {key: value for key, value in payload.items() if key not in omitted}
+    view["result_delivery_error"] = {
+        "error_code": error.error_code,
+        "reason": str(error),
+        "recommended_action": "Preserve the original result controls and Turn identity. "
+            "Repair capture privacy or remove --max-page-output-bytes and read back the "
+            "same operation; do not repeat admission or assume it did not commit.",
+        "omitted_diagnostics": omitted,
+    }
+    if (view.get("ok") is True and (view.get("appended") or view.get("idempotent_replay"))
+            and view.get("json_path") and "quota_event" in view
+            and len(_native_output_bytes(view)) + len(os.linesep) - 1 > output_bytes):
+        # The durable spend record already contains the event and its snapshots.
+        view.pop("quota_event")
+        omitted.append("quota_event")
+    return view
+
+
 def budgeted_settlement_result(
     payload: dict[str, object], *, runtime_root: Path, output_bytes: int,
     reader_argv: list[str] | None = None,

tests/control_plane/test_quota_result_budget.py

@@ -10,7 +10,7 @@
 
 import pytest
 
-from loopx.cli_commands.quota_capture import budgeted_settlement_result
+from loopx.cli_commands.quota_capture import budgeted_settlement_result, settlement_capture_failure_view
 from loopx.control_plane.quota.error_codes import QuotaDecisionTransportError
 
 
@@ -101,6 +101,27 @@ def test_reference_without_posix_identity_fails_structurally_before_open(monkeyp
         decision_capture.read_context_read_reference(value)
 
 
+@pytest.mark.parametrize("replayed", [False, True])
+def test_failed_capture_omits_large_durable_event_and_preserves_receipt(tmp_path, replayed):
+    event = {"before": "漢字" * 2000, "after": {"spent_slots": 1}}
+    artifact = tmp_path / "spend.json"
+    artifact.write_text(json.dumps({"quota_event": event}), encoding="utf-8")
+    identity = {"turn_instance_id": "original-turn"}
+    receipts = {"ok": True, "receipts": [{"step_kind": "quota_spend"}]}
+    payload = {"ok": True, "appended": not replayed, "idempotent_replay": replayed, "quota_event": event,
+        "before": event["before"], "after": event["after"], "json_path": str(artifact),
+        "settlement_identity": identity, "settlement_result": receipts, "future_control": {"keep": True}}
+    error = QuotaDecisionTransportError("Private capture directory failed validation")
+    view = settlement_capture_failure_view(payload, output_bytes=8192, error=error)
+    assert len(json.dumps(view, ensure_ascii=False, separators=(",", ":")).encode()) + len(os.linesep) <= 8192
+    assert view["settlement_result"] == receipts and view["settlement_identity"] == identity
+    assert view["idempotent_replay"] is replayed and view["appended"] is not replayed
+    assert view["future_control"] == payload["future_control"]
+    assert "quota_event" not in view
+    assert view["result_delivery_error"]["omitted_diagnostics"] == ["before", "after", "quota_event"]
+    assert json.loads(Path(view["json_path"]).read_text(encoding="utf-8"))["quota_event"] == payload["quota_event"]
+
+
 def test_large_result_without_getuid_preserves_private_boundary(tmp_path, monkeypatch):
     from loopx.control_plane.turn_driver import decision_capture