fix(quota): bound model-facing settlement diagnostics

docs/reference/required-work-context.md

@@ -1084,6 +1084,50 @@ 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. 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.
+
+生成的 Turn 绑定 JSON `quota spend-slot` 命令携带
+`--max-page-output-bytes 64000`。此输出选项不启用 Context Delivery，也不建立原生
+连续性基线。小结果保持完整；超出预算时，将 `before` / `after` 快照存入私有、
+不可变的完整 JSON，其他字段仍直接返回，包括实际追加结果、工作目录拒绝和结算
+回执。按需在 `result_details.command` 后添加 `--select`；读取器保留相同输出预算，
+每页校验完整摘要。如果其余控制信息也超限，返回命令会选择全部控制字段，必须
+完整读完这些页再行动，不截断控制正文。
+
+省略选项保留原有完整结果 API；非正预算在执行前拒绝。缺少已验证 POSIX 私有捕获
+校验的平台在结算前拒绝此选项；仅移除输出选项即可使用同一 Turn 的完整结果 API。
+reader reference 在该平台也安全拒绝，不能用 mode bits 代替 Windows ACL 校验。
+求值后捕获失败时保留全部原始控制字段（包括 `ok`、拒绝原因、身份和回执），只省略
+`before`／`after`，追加 `result_delivery_error` 并非零退出。此故障兜底优先保全控制
+信息，不保证展示预算；被拒绝的操作不一定存在 spend receipt。修复捕获权限或以
+原身份读取无输出预算的结果，不能据投递失败认定效果未提交。恢复不重新准入或
+更换原 Turn／effect 身份。此传输变化本身不证明模型保留或成本收益。
+
 ### Saved context cache / 已保存上下文缓存
 
 For `quota context-project`, enabling context delivery no longer requires a

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,25 @@ 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:
+            # 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))
+        return 0 if payload.get("ok") is True else 1
     try:
         context_delivery_enabled = native_context_activation(args)["enabled"]
     except QuotaCommandValidationError:

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,
 )
 
 
@@ -753,6 +753,74 @@ 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:
+        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 != 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))
+        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 +942,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 +961,17 @@ 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:
+        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
+    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)

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=(

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,

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,14 +273,15 @@ 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():
         raise ValueError("reader reference must be a physical file, not a symlink")
     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: