From 6808cd33adbb4ed42e4f21e1a2557bb5063f5cc7 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Mon, 2 Nov 2026 20:21:12 +0530
Subject: [PATCH 1/2] fix(quota): bound model-facing settlement diagnostics

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 docs/reference/required-work-context.md       | 33 +++++++
 loopx/cli_commands/quota.py                   | 20 +++-
 loopx/cli_commands/quota_capture.py           | 78 +++++++++++++++-
 loopx/cli_commands/quota_registration.py      |  5 +-
 loopx/control_plane/quota/effect_program.py   |  3 +-
 .../control_plane/test_quota_result_budget.py | 92 +++++++++++++++++++
 .../test_quota_settlement_cli.py              | 62 ++++++++++++-
 7 files changed, 282 insertions(+), 11 deletions(-)
 create mode 100644 tests/control_plane/test_quota_result_budget.py

diff --git a/docs/reference/required-work-context.md b/docs/reference/required-work-context.md
index f48a530000..65ae99e656 100644
--- a/docs/reference/required-work-context.md
+++ b/docs/reference/required-work-context.md
@@ -1084,6 +1084,39 @@ pointer 从原始摘要绑定的 capture 批量读取详情。提供已有 `--ca
 包括根选择在内的目录/详情 EOF 不证明全量模型保留；原生文件比较与此独立。
 模型可直接查当前完整文件，也可查基线和差分，无须先通读。
 
+### Bounded settlement results / 有界结算结果
+
+Generated Turn-bound JSON `quota spend-slot` commands include
+`--max-page-output-bytes 64000`. This output option does not enable Context
+Delivery or create a native continuity baseline. Small results remain complete;
+when a result exceeds the budget, its `before` / `after` snapshots move to a
+private immutable JSON capture. The other fields, including the actual append
+outcome, workspace refusal and settlement receipts, remain visible. Use
+`result_details.command` with `--select` for needed diagnostic facts; the reader
+keeps the same output budget and verifies the full capture digest on every page.
+If the remaining controls also exceed the limit, the returned command selects
+all of them: read those pages completely before acting. No control text is cut.
+
+Omitting the option preserves the existing full-result API. A nonpositive
+budget is rejected before execution. A capture or metadata delivery failure
+after evaluation reports the actual settlement outcome and requires recovery
+from the original receipt; it does not prove that the effect failed to commit.
+Never rerun admission or change the original Turn/effect identity to recover
+output. This transport change alone does not prove model retention or savings.
+
+生成的 Turn 绑定 JSON `quota spend-slot` 命令携带
+`--max-page-output-bytes 64000`。此输出选项不启用 Context Delivery，也不建立原生
+连续性基线。小结果保持完整；超出预算时，将 `before` / `after` 快照存入私有、
+不可变的完整 JSON，其他字段仍直接返回，包括实际追加结果、工作目录拒绝和结算
+回执。按需在 `result_details.command` 后添加 `--select`；读取器保留相同输出预算，
+每页校验完整摘要。如果其余控制信息也超限，返回命令会选择全部控制字段，必须
+完整读完这些页再行动，不截断控制正文。
+
+省略选项保留原有完整结果 API；非正预算在执行前拒绝。求值后 capture 或元数据
+投递失败时，错误仍报告实际结算结果，并要求从原回执恢复，不能据此认定效果未
+提交。恢复输出不重新准入，也不更换原 Turn／effect 身份。此传输变化本身不证明
+模型保留或成本收益。
+
 ### Saved context cache / 已保存上下文缓存
 
 For `quota context-project`, enabling context delivery no longer requires a
diff --git a/loopx/cli_commands/quota.py b/loopx/cli_commands/quota.py
index 8a6bd12e08..24d61db080 100644
--- a/loopx/cli_commands/quota.py
+++ b/loopx/cli_commands/quota.py
@@ -13,7 +13,7 @@
     _quota_renderer as _quota_renderer,
 )
 from .quota_capture import (
-    bind_capture_selection_transport, capture_decision, handoff_decision,
+    bind_capture_selection_transport, budgeted_settlement_result, capture_decision, handoff_decision,
     handle_native_context_command, handle_replan_view, prepare_quota_capture, native_context_capture_view,
     native_context_activation, model_context_reference,
 )
@@ -337,6 +337,24 @@ def _emit_quota_result(
             turn_id=_effective_spend_turn_instance_id(payload, heartbeat_turn_id=heartbeat_turn_id),
             started_at=usage_quota_started,
         )
+    if (args.quota_command == "spend-slot" and args.format == "json" and context is not None
+            and getattr(args, "max_page_output_bytes", None) is not None):
+        try:
+            view = budgeted_settlement_result(payload, runtime_root=context.runtime_root,
+                output_bytes=args.max_page_output_bytes)
+        except QuotaDecisionTransportError as exc:
+            view = {"ok": False, "mode": args.quota_command, "goal_id": args.goal_id,
+                "error_code": exc.error_code, "reason": str(exc),
+                "recommended_action": "Result delivery failed after the spend operation was evaluated. "
+                    "Read its original settlement receipt before recovery; do not repeat admission "
+                    "or assume the operation did not commit."}
+            for field in ("appended", "registry_mutated", "settlement", "settlement_progress", "heartbeat_receipt"):
+                if field in payload:
+                    view[field] = payload[field]
+            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))
+        return 0 if payload.get("ok") is True else 1
     try:
         context_delivery_enabled = native_context_activation(args)["enabled"]
     except QuotaCommandValidationError:
diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py
index a635ec1122..4f121017a6 100644
--- a/loopx/cli_commands/quota_capture.py
+++ b/loopx/cli_commands/quota_capture.py
@@ -753,6 +753,73 @@ def capture_decision(directory: Path | None, payload: dict[str, object]) -> None
         ) from exc
 
 
+def budgeted_settlement_result(
+    payload: dict[str, object], *, runtime_root: Path, output_bytes: int,
+) -> dict[str, object]:
+    """Present one evaluated result with its complete immutable diagnostic.
+
+    This is output IO, not native continuity or another quota decision.
+    Unbudgeted callers retain their existing full result.
+    """
+    from ..control_plane.effect_runtime import MAX_REQUEST_BYTES
+    from ..control_plane.turn_driver.decision_capture import write_decision_capture
+
+    def fits(value: dict[str, object]) -> bool:
+        # Match the JSON CLI serializer, including print's newline.
+        return len(json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= output_bytes
+
+    if fits(payload):
+        return payload
+    try:
+        root = runtime_root.expanduser().resolve() / "context-delivery"
+        root.mkdir(mode=0o700, exist_ok=True)
+        metadata = root.lstat()
+        if (not stat.S_ISDIR(metadata.st_mode) or metadata.st_uid != os.getuid()
+                or metadata.st_mode & 0o077):
+            raise ValueError("result capture requires a private physical directory owned by this user")
+        directory = Path(tempfile.mkdtemp(prefix="settlement-", dir=root))
+        saved = directory / "result.json"
+        write_decision_capture(saved, payload)
+        raw = saved.read_bytes()
+        if len(raw) > MAX_REQUEST_BYTES:
+            raise ValueError(f"complete result exceeds the existing reader limit: {saved}")
+        digest = hashlib.sha256(raw).hexdigest()
+        reference = write_context_read_reference(saved, digest)
+        details = {"reference_file": str(reference),
+            "command": context_capture_read_command(saved, digest, reference_file=reference,
+                page_bytes=min(NATIVE_CONTEXT_PAGE_BYTES, output_bytes),
+                max_page_output_bytes=output_bytes)}
+        deferred = [key for key in ("before", "after") if key in payload]
+        view = {key: value for key, value in payload.items() if key not in deferred}
+        view["result_details"] = {**details, "deferred_fields": deferred}
+        view["instruction"] = (
+            "The operation was already evaluated. before/after are complete in the immutable result; "
+            "read only needed details with context-read --select. Preserve the original Turn and effect "
+            "identity during recovery; do not repeat admission or infer authority from this presentation."
+        )
+        if fits(view) and "result_details" not in payload and "instruction" not in payload:
+            return view
+        # 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]
+        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),
+            "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. before/after remain available in the same immutable result."}
+        if fits(view):
+            return view
+        raise ValueError(f"result reader metadata exceeds the output budget; complete result: {saved}")
+    except (OSError, ValueError) as exc:
+        raise QuotaDecisionTransportError(
+            "The quota operation was already evaluated, but bounded result delivery failed. "
+            "Recover the original operation/receipt; do not repeat admission or assume it did not commit. "
+            f"{exc}"
+        ) from exc
+
+
 def bind_capture_selection_transport(payload: dict[str, Any], args: argparse.Namespace, *,
                                      capture_root: Path | None = None) -> None:
     """Retain caller display/capture options in the existing selection command.
@@ -874,9 +941,11 @@ def handle_native_context_command(args: argparse.Namespace, *, print_payload: Ca
         print_payload({"ok": False, "reason": "--list is only supported by context-read"}, args.format, str)
         return 1
     has_output_limit = getattr(args, "max_page_output_bytes", None) is not None
+    settlement_output = (args.quota_command == "spend-slot" and args.format == "json"
+        and bool(getattr(args, "turn_instance_id", None)) and bool(getattr(args, "agent_id", None)))
     budget_scope_error = ("max-page-output-bytes is only supported by JSON context-read "
-        "or enabled native should-run/context-project")
-    if has_output_limit and args.quota_command != "context-read" and (
+        "or enabled native should-run/context-project, or Turn-bound JSON spend-slot")
+    if has_output_limit and args.quota_command != "context-read" and not settlement_output and (
         args.quota_command not in ("should-run", "context-project") or args.format != "json"
         or getattr(args, "context_delivery", None) == "off"):
         print_payload({"ok": False, "reason": budget_scope_error},
@@ -891,7 +960,10 @@ def handle_native_context_command(args: argparse.Namespace, *, print_payload: Ca
             print_payload({"ok": False, "reason": str(exc)}, args.format,
                 lambda value: json.dumps(value, ensure_ascii=False, indent=2))
             return 1
-    if has_output_limit and args.quota_command != "context-read":
+    if has_output_limit and settlement_output and args.max_page_output_bytes <= 0:
+        print_payload({"ok": False, "reason": "max-page-output-bytes must be positive"}, args.format, str)
+        return 1
+    if has_output_limit and args.quota_command != "context-read" and not settlement_output:
         try:
             if not native_context_activation(args)["enabled"]:
                 raise QuotaCommandValidationError(budget_scope_error)
diff --git a/loopx/cli_commands/quota_registration.py b/loopx/cli_commands/quota_registration.py
index d6ce128f8d..bed2a3411c 100644
--- a/loopx/cli_commands/quota_registration.py
+++ b/loopx/cli_commands/quota_registration.py
@@ -249,8 +249,9 @@ def register_quota_command(
              "Larger pages require a sufficient --max-page-output-bytes within the capture limit; "
              "use only with known sufficient direct and retained-history output budgets.")
     quota_parser.add_argument("--max-page-output-bytes", type=int,
-        help="For JSON context-read and enabled native should-run/context-project: optional positive byte limit for successful page stdout, including JSON framing and newline. "
-             "Content shrinks at UTF-8 boundaries; next_command preserves the limit. An insufficient limit returns an error.")
+        help="For JSON context-read, enabled native should-run/context-project, or Turn-bound JSON spend-slot: optional positive output byte limit. "
+             "Reader pages shrink at UTF-8 boundaries; oversized settlement diagnostics remain complete in a private immutable result with reader references. "
+             "An insufficient metadata budget returns an error; this option does not enable native context continuity.")
     quota_parser.add_argument(
         "--turn-instance-id",
         help=(
diff --git a/loopx/control_plane/quota/effect_program.py b/loopx/control_plane/quota/effect_program.py
index f9247073b5..bdf24c7680 100644
--- a/loopx/control_plane/quota/effect_program.py
+++ b/loopx/control_plane/quota/effect_program.py
@@ -29,6 +29,7 @@
 )
 from ..goals.goal_ref_validation import exact_goal_ref
 from ..work_items.delivery_batch_scale import DELIVERY_BATCH_SCALE_INPUT_CHOICES
+from ..turn_driver.decision_capture import NATIVE_CONTEXT_PAGE_OUTPUT_BYTES
 
 __all__ = [
     "SETTLEMENT_IDENTITY_SCHEMA_VERSION",
@@ -242,7 +243,7 @@ def build_turn_scoped_cli_settlement_plan(
     spend = (
         f"{cli_prefix} quota spend-slot --goal-id {shlex.quote(goal_id)} --slots 1 "
         f"--source {quota_spend_source} --execute{binding_arg}{turn_arg}{goal_ref_arg}"
-        f"{scoped_cli_args}"
+        f"{scoped_cli_args} --max-page-output-bytes {NATIVE_CONTEXT_PAGE_OUTPUT_BYTES}"
     )
     payload = effect_runtime_result("settlement.turn_scoped_cli_plan", {
         "identity": identity.as_dict(), "delivery_boundary": delivery_boundary,
diff --git a/tests/control_plane/test_quota_result_budget.py b/tests/control_plane/test_quota_result_budget.py
new file mode 100644
index 0000000000..59dd0da4bb
--- /dev/null
+++ b/tests/control_plane/test_quota_result_budget.py
@@ -0,0 +1,92 @@
+"""Bounded settlement presentation preserves the original result and recovery."""
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+import shlex
+import subprocess
+import sys
+
+import pytest
+
+from loopx.cli_commands.quota_capture import budgeted_settlement_result
+from loopx.control_plane.quota.error_codes import QuotaDecisionTransportError
+
+
+def _read(command: str) -> tuple[int, dict]:
+    words = shlex.split(command)
+    result = subprocess.run([sys.executable, "-m", "loopx.cli", "--format", "json",
+        *words[words.index("quota"):]], capture_output=True, text=True, check=False,
+        cwd=Path(__file__).resolve().parents[2], env={**os.environ, "LOOPX_USAGE_PING": "0"})
+    return result.returncode, json.loads(result.stdout)
+
+
+@pytest.mark.parametrize("ok", [False, True])
+def test_small_result_stays_complete_without_capture(tmp_path, ok):
+    original = {"ok": ok, "appended": ok, "after": None, "unknown": [False, 0, "", None]}
+    assert budgeted_settlement_result(original, runtime_root=tmp_path, output_bytes=8192) == original
+    assert not (tmp_path / "context-delivery").exists()
+
+
+def test_large_diagnostic_keeps_controls_and_immutable_complete_source(tmp_path):
+    original = {"ok": False, "mode": "spend-slot", "appended": False, "registry_mutated": False,
+        "workspace_guard": {"action": "return_to_delivery_worktree", "blocks_delivery": True},
+        "before": {"text": "complete source 漢字 " * 10000, "last": "Original stop remains."},
+        "after": None, "unknown": {"value": 2**61 + 1, "empty": []}}
+    view = budgeted_settlement_result(original, runtime_root=tmp_path, output_bytes=8192)
+    for field in ("ok", "mode", "appended", "registry_mutated", "workspace_guard", "unknown"):
+        assert view[field] == original[field]
+    assert len(json.dumps(view, ensure_ascii=True, indent=2).encode()) + 1 <= 8192
+    reference = Path(view["result_details"]["reference_file"])
+    captured = Path(json.loads(reference.read_text())["path"])
+    assert json.loads(captured.read_text()) == original
+    assert captured.stat().st_mode & 0o077 == 0
+    assert reference.stat().st_mode & 0o077 == 0
+    rc, page = _read(view["result_details"]["command"] + " --select /unknown --select /workspace_guard")
+    assert rc == 0
+    selected = {row["pointer"]: row["value"] for row in json.loads(page["content"])["selections"]}
+    assert selected == {"/unknown": original["unknown"], "/workspace_guard": original["workspace_guard"]}
+    captured.write_text(json.dumps({**original, "before": {"text": "replacement"}}))
+    rc, rejected = _read(view["result_details"]["command"])
+    assert rc == 1 and rejected["ok"] is False
+    captured.unlink()
+    rc, missing = _read(view["result_details"]["command"])
+    assert rc == 1 and missing["ok"] is False
+
+
+def test_large_unknown_control_is_reachable_without_truncation(tmp_path):
+    original = {"ok": False, "future/control~v": {"stop": "must remain " * 4000 + "FINAL STOP"},
+        "instruction": "Preserve the original control instruction.", "result_details": {"future": True},
+        "before": {"history": "diagnostic " * 10000}}
+    view = budgeted_settlement_result(original, runtime_root=tmp_path, output_bytes=8192)
+    assert "command" in view
+    parts = []
+    command = view["command"]
+    while command:
+        rc, page = _read(command)
+        assert rc == 0
+        parts.append(page["content"])
+        assert len(json.dumps(page, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= 8192
+        command = page.get("next_command")
+    selected = {row["pointer"]: row["value"] for row in json.loads("".join(parts))["selections"]}
+    assert selected["/future~1control~0v"] == original["future/control~v"]
+    assert selected["/instruction"] == original["instruction"]
+    assert selected["/result_details"] == original["result_details"]
+
+
+@pytest.mark.parametrize("symlink", [False, True])
+def test_result_capture_rejects_public_or_symlink_root(tmp_path, symlink):
+    original = {"ok": False, "before": {"history": "complete " * 20000}}
+    public = tmp_path / "public"
+    public.mkdir()
+    if symlink:
+        public.chmod(0o700)
+        (tmp_path / "context-delivery").symlink_to(public, target_is_directory=True)
+    else:
+        public.rename(tmp_path / "context-delivery")
+    with pytest.raises(QuotaDecisionTransportError):
+        budgeted_settlement_result(original, runtime_root=tmp_path, output_bytes=8192)
+    target = public if symlink else tmp_path / "context-delivery"
+    assert not list(target.iterdir())
+    assert target.stat().st_mode & 0o777 == (0o700 if symlink else 0o755)
diff --git a/tests/control_plane/test_quota_settlement_cli.py b/tests/control_plane/test_quota_settlement_cli.py
index 3e9586671c..7d08312863 100644
--- a/tests/control_plane/test_quota_settlement_cli.py
+++ b/tests/control_plane/test_quota_settlement_cli.py
@@ -7445,10 +7445,12 @@ def test_receipt_review_does_not_persist_private_or_lossy_stop_clauses() -> None
 
 
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
+@pytest.mark.parametrize("bounded_result", [False, True])
 def test_peer_refresh_rejects_implicit_canonical_workspace_before_writeback(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,
     provider: str,
+    bounded_result: bool,
 ) -> None:
     if provider == "sqlite":
         from canonical_authority_fixture import isolate_sqlite_runtime
@@ -7575,10 +7577,62 @@ def test_peer_refresh_rejects_implicit_canonical_workspace_before_writeback(
     )
     assert linked["delivery_workspace"]["peer_independent_worktree_required"] is True
     command = linked["settlement_owed"]["command"]
-    spent_rc, spent = _run_generated_cli(command, registry_path=registry_path, cwd=linked_worktree)
-    assert spent_rc == 0 and spent["settlement_progress"]["state"] == "settled", spent
-    repeated_rc, repeated = _run_generated_cli(command, registry_path=registry_path, cwd=linked_worktree)
-    assert repeated_rc == 0 and repeated["appended"] is False, repeated
+    assert "--max-page-output-bytes 64000" in command
+    if bounded_result:
+        command = command.replace("--max-page-output-bytes 64000", "--max-page-output-bytes 8192")
+        # Wrong workspace remains a refusal, with no accounting effect. The
+        # full unbudgeted API remains available for independent comparison.
+        raw_command = command.replace(" --max-page-output-bytes 8192", "")
+        raw_rc, raw = _run_generated_cli(raw_command, registry_path=registry_path, cwd=project)
+        assert raw_rc == 1 and raw["appended"] is False
+        assert raw["before"]["selected_todo"]["todo_id"] == TODO_ID
+        assert len(json.dumps(raw).encode()) > 8192
+        refused_rc, refused = _run_generated_cli(command, registry_path=registry_path, cwd=project)
+        assert refused_rc == 1 and refused["workspace_guard"] == raw["workspace_guard"]
+        assert refused["appended"] is False and refused["registry_mutated"] is False
+        assert "before" not in refused and _spend_run_count(runtime) == 0
+        reference = Path(refused["result_details"]["reference_file"])
+        original = Path(json.loads(reference.read_text())["path"])
+        complete = json.loads(original.read_text())
+        assert complete["before"]["selected_todo"]["todo_id"] == TODO_ID
+        assert len(json.dumps(refused, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= 8192
+        assert original.stat().st_mode & 0o077 == 0
+        read_rc, page = _run_cli(registry_path, runtime, "quota", "context-read",
+            "--reference-file", str(reference), "--select", "/workspace_guard", cwd=project)
+        assert read_rc == 0 and page["eof"]
+        assert json.loads(page["content"])["selections"][0]["value"] == raw["workspace_guard"]
+        assert not list((runtime / "context-delivery").rglob("session.json"))
+        # Invalid presentation input is rejected before the original effect.
+        invalid_rc, invalid = _run_generated_cli(command.replace("8192", "0"),
+            registry_path=registry_path, cwd=linked_worktree)
+        assert invalid_rc == 1 and "positive" in invalid["reason"]
+        assert _spend_run_count(runtime) == 0
+        # A later IO failure cannot conceal the effect that already committed.
+        capture_root = runtime / "context-delivery"
+        capture_root.chmod(0o755)
+        failed_rc, failed = _run_generated_cli(command, registry_path=registry_path, cwd=linked_worktree)
+        assert failed_rc == 1 and failed["error_code"] == "quota_decision_transport_failed"
+        assert failed["appended"] is True and _spend_run_count(runtime) == 1
+        assert failed["settlement_progress"]["state"] == "settled"
+        assert "original settlement receipt" in failed["recommended_action"]
+        capture_root.chmod(0o700)
+    for replay in (False, True):
+        spent_rc, spent = _run_generated_cli(command, registry_path=registry_path, cwd=linked_worktree)
+        assert spent_rc == 0, spent
+        if "command" in spent:
+            parts = []
+            read_command = spent["command"]
+            while read_command:
+                read_rc, page = _run_generated_cli(read_command, registry_path=registry_path, cwd=linked_worktree)
+                assert read_rc == 0
+                parts.append(page["content"])
+                read_command = page.get("next_command")
+            spent = {row["pointer"][1:]: row["value"]
+                for row in json.loads("".join(parts))["selections"]}
+        if not replay and not bounded_result:
+            assert spent["settlement_progress"]["state"] == "settled", spent
+        else:
+            assert spent["appended"] is False, spent
     assert _spend_run_count(runtime) == 1
 
 

From 257b45585f908b1a9ec7d7d27b294a7cf8389217 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Tue, 3 Nov 2026 02:31:19 +0800
Subject: [PATCH 2/2] fix(quota): preserve evaluated controls on delivery
 failure

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 docs/reference/required-work-context.md       | 25 ++++++++++----
 loopx/cli_commands/quota.py                   | 17 +++++-----
 loopx/cli_commands/quota_capture.py           | 12 +++++--
 .../turn_driver/decision_capture.py           | 19 +++++++++--
 .../control_plane/test_quota_result_budget.py | 33 +++++++++++++++++++
 .../test_quota_settlement_cli.py              | 15 +++++++--
 6 files changed, 100 insertions(+), 21 deletions(-)

diff --git a/docs/reference/required-work-context.md b/docs/reference/required-work-context.md
index 65ae99e656..c9f2afb69f 100644
--- a/docs/reference/required-work-context.md
+++ b/docs/reference/required-work-context.md
@@ -1098,9 +1098,16 @@ If the remaining controls also exceed the limit, the returned command selects
 all of them: read those pages completely before acting. No control text is cut.
 
 Omitting the option preserves the existing full-result API. A nonpositive
-budget is rejected before execution. A capture or metadata delivery failure
-after evaluation reports the actual settlement outcome and requires recovery
-from the original receipt; it does not prove that the effect failed to commit.
+budget is rejected before execution. Platforms without qualified POSIX private
+capture validation reject this option before settlement; remove only the output
+option to use the same Turn's full-result API. Reader references also fail closed
+there; mode bits are not a substitute for Windows ACL verification.
+A capture failure after evaluation keeps every original control field (including
+`ok`, refusal reason, identity and receipts), omits only `before` / `after`, and
+adds `result_delivery_error`; the process exits nonzero. This fallback prioritizes
+complete controls over the display budget. A refused operation need not have a
+spend receipt. Repair capture privacy or read the same operation without the
+output option; a delivery failure does not prove that the effect failed to commit.
 Never rerun admission or change the original Turn/effect identity to recover
 output. This transport change alone does not prove model retention or savings.
 
@@ -1112,10 +1119,14 @@ output. This transport change alone does not prove model retention or savings.
 每页校验完整摘要。如果其余控制信息也超限，返回命令会选择全部控制字段，必须
 完整读完这些页再行动，不截断控制正文。
 
-省略选项保留原有完整结果 API；非正预算在执行前拒绝。求值后 capture 或元数据
-投递失败时，错误仍报告实际结算结果，并要求从原回执恢复，不能据此认定效果未
-提交。恢复输出不重新准入，也不更换原 Turn／effect 身份。此传输变化本身不证明
-模型保留或成本收益。
+省略选项保留原有完整结果 API；非正预算在执行前拒绝。缺少已验证 POSIX 私有捕获
+校验的平台在结算前拒绝此选项；仅移除输出选项即可使用同一 Turn 的完整结果 API。
+reader reference 在该平台也安全拒绝，不能用 mode bits 代替 Windows ACL 校验。
+求值后捕获失败时保留全部原始控制字段（包括 `ok`、拒绝原因、身份和回执），只省略
+`before`／`after`，追加 `result_delivery_error` 并非零退出。此故障兜底优先保全控制
+信息，不保证展示预算；被拒绝的操作不一定存在 spend receipt。修复捕获权限或以
+原身份读取无输出预算的结果，不能据投递失败认定效果未提交。恢复不重新准入或
+更换原 Turn／effect 身份。此传输变化本身不证明模型保留或成本收益。
 
 ### Saved context cache / 已保存上下文缓存
 
diff --git a/loopx/cli_commands/quota.py b/loopx/cli_commands/quota.py
index 24d61db080..2f5afb956b 100644
--- a/loopx/cli_commands/quota.py
+++ b/loopx/cli_commands/quota.py
@@ -343,14 +343,15 @@ def _emit_quota_result(
             view = budgeted_settlement_result(payload, runtime_root=context.runtime_root,
                 output_bytes=args.max_page_output_bytes)
         except QuotaDecisionTransportError as exc:
-            view = {"ok": False, "mode": args.quota_command, "goal_id": args.goal_id,
-                "error_code": exc.error_code, "reason": str(exc),
-                "recommended_action": "Result delivery failed after the spend operation was evaluated. "
-                    "Read its original settlement receipt before recovery; do not repeat admission "
-                    "or assume the operation did not commit."}
-            for field in ("appended", "registry_mutated", "settlement", "settlement_progress", "heartbeat_receipt"):
-                if field in payload:
-                    view[field] = payload[field]
+            # 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]}
             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))
diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py
index 4f121017a6..97d09ac7ed 100644
--- a/loopx/cli_commands/quota_capture.py
+++ b/loopx/cli_commands/quota_capture.py
@@ -21,7 +21,7 @@
     NATIVE_CONTEXT_PAGE_BYTES, NATIVE_CONTEXT_PAGE_OUTPUT_BYTES, CONTEXT_CAPTURE_NAVIGATION_GUIDANCE,
     native_context_page_budget,
     read_context_capture_file as _read_capture,
-    read_context_read_reference, write_context_read_reference,
+    read_context_read_reference, write_context_read_reference, private_capture_uid,
 )
 
 
@@ -771,10 +771,11 @@ def fits(value: dict[str, object]) -> bool:
     if fits(payload):
         return payload
     try:
+        uid = private_capture_uid()
         root = runtime_root.expanduser().resolve() / "context-delivery"
         root.mkdir(mode=0o700, exist_ok=True)
         metadata = root.lstat()
-        if (not stat.S_ISDIR(metadata.st_mode) or metadata.st_uid != os.getuid()
+        if (not stat.S_ISDIR(metadata.st_mode) or metadata.st_uid != uid
                 or metadata.st_mode & 0o077):
             raise ValueError("result capture requires a private physical directory owned by this user")
         directory = Path(tempfile.mkdtemp(prefix="settlement-", dir=root))
@@ -960,6 +961,13 @@ def handle_native_context_command(args: argparse.Namespace, *, print_payload: Ca
             print_payload({"ok": False, "reason": str(exc)}, args.format,
                 lambda value: json.dumps(value, ensure_ascii=False, indent=2))
             return 1
+    if has_output_limit and settlement_output:
+        try:
+            private_capture_uid()
+        except ValueError as exc:
+            print_payload({"ok": False, "reason": str(exc), "appended": False,
+                "registry_mutated": False}, args.format, str)
+            return 1
     if has_output_limit and settlement_output and args.max_page_output_bytes <= 0:
         print_payload({"ok": False, "reason": "max-page-output-bytes must be positive"}, args.format, str)
         return 1
diff --git a/loopx/control_plane/turn_driver/decision_capture.py b/loopx/control_plane/turn_driver/decision_capture.py
index 086b0ccb52..c407678d44 100644
--- a/loopx/control_plane/turn_driver/decision_capture.py
+++ b/loopx/control_plane/turn_driver/decision_capture.py
@@ -18,6 +18,19 @@
 from ..content_digest import BARE_SHA256_PATTERN
 
 
+def private_capture_uid() -> int:
+    """Fail closed until this IO boundary has a qualified non-POSIX ACL reader.
+
+    POSIX mode bits cannot establish Windows privacy. Check before creating or
+    reading private captures, and before evaluating bounded settlement commands.
+    """
+    getuid = getattr(os, "getuid", None)
+    if os.name != "posix" or not callable(getuid):
+        raise ValueError("private capture validation is unsupported on this platform; "
+                         "use the same Turn command without --max-page-output-bytes")
+    return getuid()
+
+
 def write_decision_capture(
     path: Path, payload: Mapping[str, Any], *, publish: Callable[[str, Path], None] = os.link,
 ) -> None:
@@ -260,6 +273,7 @@ def read_context_read_reference(value: str) -> tuple[Path, dict[str, Any]]:
     """Read one private physical descriptor; the caller verifies capture bytes."""
     from ..effect_runtime import MAX_REQUEST_BYTES
 
+    uid = private_capture_uid()
     path = Path(value).expanduser().absolute()
     path = path.parent.resolve(strict=True) / path.name
     if path.is_symlink():
@@ -267,7 +281,7 @@ def read_context_read_reference(value: str) -> tuple[Path, dict[str, Any]]:
     descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0))
     with os.fdopen(descriptor, "rb") as stream:
         info = os.fstat(stream.fileno())
-        if (not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid() or info.st_mode & 0o077
+        if (not stat.S_ISREG(info.st_mode) or info.st_uid != uid or info.st_mode & 0o077
                 or info.st_size > MAX_REQUEST_BYTES):
             raise ValueError("reader reference must be a private regular file owned by this user")
         raw = stream.read(MAX_REQUEST_BYTES + 1)
@@ -288,6 +302,7 @@ def read_context_capture_file(value: str, digest: str, *, private: bool = False)
 
     if not isinstance(digest, str) or BARE_SHA256_PATTERN.fullmatch(digest) is None or not value:
         raise ValueError("context-read requires the original capture file and its exact SHA-256")
+    uid = private_capture_uid() if private else None
     path = Path(value).expanduser().absolute()
     # Parent aliases such as macOS /tmp are normal capture locations.
     # Resolve those once, but never follow a substituted final file.
@@ -299,7 +314,7 @@ def read_context_capture_file(value: str, digest: str, *, private: bool = False)
         metadata = os.fstat(stream.fileno())
         if not stat.S_ISREG(metadata.st_mode) or metadata.st_size > MAX_REQUEST_BYTES:
             raise ValueError("capture must be a regular JSON file within the existing transport size limit")
-        if private and (metadata.st_uid != os.getuid() or metadata.st_mode & 0o077):
+        if private and (metadata.st_uid != uid or metadata.st_mode & 0o077):
             raise ValueError("native context requires a private capture owned by this user")
         raw = stream.read(metadata.st_size + 1)
     if len(raw) != metadata.st_size or hashlib.sha256(raw).hexdigest() != digest:
diff --git a/tests/control_plane/test_quota_result_budget.py b/tests/control_plane/test_quota_result_budget.py
index 59dd0da4bb..5326f0fe39 100644
--- a/tests/control_plane/test_quota_result_budget.py
+++ b/tests/control_plane/test_quota_result_budget.py
@@ -90,3 +90,36 @@ def test_result_capture_rejects_public_or_symlink_root(tmp_path, symlink):
     target = public if symlink else tmp_path / "context-delivery"
     assert not list(target.iterdir())
     assert target.stat().st_mode & 0o777 == (0o700 if symlink else 0o755)
+
+
+@pytest.mark.parametrize("value", ["reader.json", r"C:\private\result.reader.json"])
+def test_reference_without_posix_identity_fails_structurally_before_open(monkeypatch, value):
+    from loopx.control_plane.turn_driver import decision_capture
+
+    monkeypatch.delattr(decision_capture.os, "getuid", raising=False)
+    with pytest.raises(ValueError, match="private.*unsupported"):
+        decision_capture.read_context_read_reference(value)
+
+
+def test_large_result_without_getuid_preserves_private_boundary(tmp_path, monkeypatch):
+    from loopx.control_plane.turn_driver import decision_capture
+
+    monkeypatch.delattr(decision_capture.os, "getuid", raising=False)
+    with pytest.raises(QuotaDecisionTransportError, match="private.*unsupported"):
+        budgeted_settlement_result({"ok": True, "before": "large" * 20000},
+            runtime_root=tmp_path, output_bytes=8192)
+    assert not (tmp_path / "context-delivery").exists()
+
+
+def test_bounded_spend_without_getuid_rejects_before_effect(tmp_path, monkeypatch):
+    from loopx.cli import main
+    from loopx.cli_commands import quota
+    from loopx.control_plane.turn_driver import decision_capture
+
+    monkeypatch.delattr(decision_capture.os, "getuid", raising=False)
+    def forbidden(*args, **kwargs):
+        pytest.fail("unsupported private capture reached effect preparation")
+    monkeypatch.setattr(quota, "_prepare_quota_command_execution", forbidden)
+    assert main(["--format", "json", "--registry", str(tmp_path / "unused.json"),
+        "quota", "spend-slot", "--goal-id", "fixture", "--agent-id", "fixture",
+        "--turn-instance-id", "same-turn", "--execute", "--max-page-output-bytes", "64000"]) == 1
diff --git a/tests/control_plane/test_quota_settlement_cli.py b/tests/control_plane/test_quota_settlement_cli.py
index 7d08312863..7b1d4cbd47 100644
--- a/tests/control_plane/test_quota_settlement_cli.py
+++ b/tests/control_plane/test_quota_settlement_cli.py
@@ -7610,11 +7610,22 @@ def test_peer_refresh_rejects_implicit_canonical_workspace_before_writeback(
         # A later IO failure cannot conceal the effect that already committed.
         capture_root = runtime / "context-delivery"
         capture_root.chmod(0o755)
+        rejected_rc, rejected = _run_generated_cli(command, registry_path=registry_path, cwd=project)
+        assert rejected_rc == 1 and rejected["appended"] is False
+        assert _spend_run_count(runtime) == 0
+        for field in ("ok", "workspace_guard", "reason", "recommended_action", "error_code"):
+            if field in raw:
+                assert rejected[field] == raw[field]
+        assert rejected["result_delivery_error"]["error_code"] == "quota_decision_transport_failed"
         failed_rc, failed = _run_generated_cli(command, registry_path=registry_path, cwd=linked_worktree)
-        assert failed_rc == 1 and failed["error_code"] == "quota_decision_transport_failed"
+        assert failed_rc == 1 and failed["ok"] is True
+        assert failed["result_delivery_error"]["error_code"] == "quota_decision_transport_failed"
+        assert failed["settlement_result"]["ok"] is True
+        assert "quota_spend" in {row["step_kind"] for row in failed["settlement_result"]["receipts"]}
+        assert failed["settlement_identity"]["turn_instance_id"] == TURN_ID
         assert failed["appended"] is True and _spend_run_count(runtime) == 1
         assert failed["settlement_progress"]["state"] == "settled"
-        assert "original settlement receipt" in failed["recommended_action"]
+        assert "original" in failed["result_delivery_error"]["recommended_action"]
         capture_root.chmod(0o700)
     for replay in (False, True):
         spent_rc, spent = _run_generated_cli(command, registry_path=registry_path, cwd=linked_worktree)