feat(context): scope native quota outcomes with same-capture recovery

docs/reference/scoped-outcome-context.md

@@ -0,0 +1,100 @@
+# Scoped outcome context in native quota delivery
+
+This is an opt-in presentation of an immutable quota capture, not a partial
+canonical decision, a new acceptance owner or the controller/executor runtime.
+It extracts the outcome-reading slice of the [unified-contract proposal](../architecture/rfcs/controller-executor-unified-contract-v0.md)
+without adopting its complete-Turn schema or changing ordinary TurnEnvelope.
+
+## Activation and rollback
+
+Use the existing [Codex differential-context setup](../heartbeat-automation-prompt.md#opt-in-native-context-delivery)
+with `--context-delivery on`. The native lifecycle hooks must actually be loaded
+and trusted; saving a preference is not evidence that a session is using it.
+The new scope is available for ordinary captured quota delivery only. An explicit
+`--turn-envelope` keeps its existing full-source behavior. Unsupported hosts,
+missing native setup and the default `off` path do not acquire a scope plan.
+There is no second switch or benchmark-specific requirement owner.
+
+For a new run, select `--context-delivery off` to restore ordinary full delivery.
+Do not change the treatment of an active or frozen trial. Before downgrading the
+reader, finish pending work or recover its original complete capture. Preserve
+both the original source and any required-context artifacts needed by an existing
+session; historical captures are never rewritten into the new format.
+
+## What is read
+
+The existing acceptance owner computes the read plan from one complete canonical
+work snapshot, with its provider revision and complete acceptance digest. It
+keeps all outcomes owned by the current Agent and the closure of the selected
+work's registered relations. Shared acceptance bindings, requests and evidence,
+including failed or stale observations, expand the required set. Ownership alone
+is not proof that another outcome is irrelevant.
+
+Only unrelated peer outcomes outside that declared closure can be deferred.
+Their current identity, revision, work links and requirement-state index remain
+visible. Their exact complete bodies remain in the original immutable capture.
+An Agent must read that detail before relying on it, switching into related work
+or assessing broader acceptance. The index is not an acceptance verdict and does
+not turn an Agent's review into independently verified success.
+
+Original Goal intent, global acceptance facts, permissions, stops, selected work,
+settlement identity and current controls remain in the presentation. The existing
+interaction owner changes only its exact fulfilled outcome-read instruction.
+Unknown guidance, any outstanding required read, incomplete work context, broad
+replanning, unbound work, missing relations, mismatched source revisions or
+unsupported outcome fields keep complete source delivery. A new relationship or
+changed source is evaluated by a fresh guard, not inferred from an old index.
+This conservative declared-relation closure does not establish semantic
+independence of arbitrary natural-language requirements.
+
+The scope is used only when the complete derived presentation, including its
+index, references and instructions, is smaller in UTF-8 bytes than the original.
+Small sources can therefore remain full even with a valid read plan. This is a
+transport-byte check, not a token-price or complete-request cost estimate.
+
+## Detail and recovery
+
+The model sees `outcome_context_projection_v0`: the required context, current
+outcome index and an exact `source_capture` path/SHA-256. With that original
+reference, the existing reader can return the index or one complete outcome:
+
+```sh
+loopx --format json quota context-read \
+  --capture-file "$CAPTURE" --capture-sha256 "$SHA256" --outcome-ref
+
+loopx --format json quota context-read \
+  --capture-file "$CAPTURE" --capture-sha256 "$SHA256" --outcome-ref "$OUTCOME_REF"
+```
+
+Follow the returned `next_command` for every page. An outcome reference identifies
+its complete captured content, not a live lookup by outcome id. Selected pages
+also pin a `selection_sha256`, because a changed rendering cannot safely reuse
+byte cursors merely because the source file has the same hash. Reading an index
+or detail does not count as reading the complete context and cannot establish a
+full-context baseline. Invalid selections retain a complete-capture fallback.
+
+The full original source is published first. The scoped presentation has its own
+immutable `required-context.json`, recovery reference and full/delta baseline;
+optional unread bodies are never silently marked as retained. If this second
+publication fails, the caller returns the complete original source. Compaction,
+lost continuity, damaged or truncated pages require recovery of the same
+captured presentation before dependent work. Recovery never runs a replacement
+guard, renews authority or spends quota. Current cancellation and admission rules
+continue to be checked by their existing owners.
+
+## Qualification boundary
+
+Focused tests exercise the TypeScript owner, actual File authority, public quota
+and reader CLI, byte-bound details, synthetic native lifecycle events, exact
+delta reconstruction and publication failure. Synthetic native events are not
+an installed-Codex or model-retention qualification. Supported SQLite,
+PostgreSQL where applicable, full repository checks and native-host qualification
+remain separate gates. The PR records which checks actually ran and any failures.
+
+A smaller capture does not establish lower complete-request tokens, cached-token
+costs, billable savings or better accepted outcomes. Measure cold delivery,
+subsequent wakes, detail reads, compaction and recovery together on a fixed source
+revision. This slice does not narrow independent outcomes owned by the same Agent
+and therefore does not claim to solve all single-Agent benchmark overhead. It
+also does not complete C1, dual controller/executor sessions or the full C2
+cross-entrypoint migration.

loopx/cli_commands/quota.py

@@ -638,6 +638,10 @@ def handle_quota_command(
                 turn_instance_id=heartbeat_turn_id,
                 interaction_projection_hooks=interaction_projection_hooks,
                 turn_start_hook_dispatch=turn_start_hook_dispatch,
+                context_read_scope=(
+                    capture_directory is not None and getattr(args, "context_delivery", "off") == "on"
+                    and not bool(getattr(args, "turn_envelope", False))
+                ),
                 goal_ref=goal_ref,
             )
             _attach_turn_start_hook_dispatch(payload, turn_start_hook_dispatch)

loopx/cli_commands/quota_capture.py

@@ -14,7 +14,7 @@
 from typing import Any
 
 from ..control_plane.content_digest import BARE_SHA256_PATTERN
-from ..control_plane.effect_runtime import EffectRuntimeRejected
+from ..control_plane.effect_runtime import EffectRuntimeRejected, EffectRuntimeRemoteError, EffectRuntimeStartupError
 from ..control_plane.quota.error_codes import QuotaCommandValidationError, QuotaDecisionTransportError
 from ..control_plane.turn_driver.decision_capture import (
     validate_handoff_socket, CONTEXT_CAPTURE_READ_GUIDANCE, CONTEXT_CAPTURE_PAGE_GUIDANCE, context_capture_read_command,
@@ -37,7 +37,9 @@ def _capture_page(path: Path, raw: bytes, digest: str, *, offset: int, size: int
                   output_limit: int | None, instruction: bool,
                   presentation: dict[str, Any] | None = None,
                   next_page_bytes: int | None = None,
-                  next_page_output_bytes: int | None = None) -> dict[str, Any]:
+                  next_page_output_bytes: int | None = None,
+                  outcome_ref: str | None = None,
+                  selection_sha256: str | None = None) -> dict[str, Any]:
     end = min(offset + size, len(raw))
     while end < len(raw) and raw[end] & 0xC0 == 0x80:
         end -= 1
@@ -48,11 +50,18 @@ def page(text: str) -> dict[str, Any]:
         result: dict[str, Any] = {**(presentation or {}), "ok": True, "schema_version": "context_capture_page_v1", "sha256": digest,
             "offset": offset, "next_offset": cursor if cursor < len(raw) else None,
             "total_bytes": len(raw), "eof": cursor == len(raw), "content": text}
+        if outcome_ref is not None:
+            result["outcome_ref"] = outcome_ref or None
+            result["selection_sha256"] = selection_sha256
+            result["read_instruction"] = (
+                "Offsets and total_bytes address only this selected outcome/index JSON; sha256 binds the entire original capture. "
+                "This read does not establish full-capture retention or replace the capture's required-read obligations.")
         if instruction:
-            result["read_instruction"] = CONTEXT_CAPTURE_PAGE_GUIDANCE
+            result["read_instruction"] = str(result.get("read_instruction", "")) + CONTEXT_CAPTURE_PAGE_GUIDANCE
         if cursor < len(raw):
             result["next_command"] = context_capture_read_command(path, digest, offset=cursor, page_bytes=next_page_bytes or size,
-                max_page_output_bytes=next_page_output_bytes or output_limit)
+                max_page_output_bytes=next_page_output_bytes or output_limit, outcome_ref=outcome_ref,
+                selection_sha256=selection_sha256)
         return result
 
     def fits(result: dict[str, Any]) -> bool:
@@ -88,6 +97,7 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An
     in that exact capture, so a replacement cannot silently mix two decisions.
     EOF describes this page, never a model's complete-consumption receipt.
     """
+    recovery_command = None
     try:
         digest = getattr(args, "capture_sha256", None)
         value = getattr(args, "capture_file", None)
@@ -124,18 +134,50 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An
         if type(offset) is not int or offset < 0 or type(size) is not int or not 256 <= size <= maximum:
             raise ValueError("context-read requires a nonnegative byte offset and page-bytes between 256 and 8192; "
                 "larger pages require a sufficient max-page-output-bytes within the capture limit")
-        path, raw, _ = _read_capture(value, digest)
+        path, raw, captured = _read_capture(value, digest)
+        outcome_ref = getattr(args, "outcome_ref", None)
+        selection_digest = None
+        expected_selection = getattr(args, "selection_sha256", None)
+        if expected_selection is not None and outcome_ref is None:
+            raise ValueError("selection-sha256 requires outcome-ref on the original context-read")
+        if outcome_ref is not None:
+            recovery_command = context_capture_read_command(path, digest)
+            # The typed owner shares exact lookup/index semantics with replan.
+            # IO and pagination remain here; no Goal/provider or guard is read.
+            from ..control_plane.quota.turn_envelope import _turn_envelope_result
+
+            source_payload = None
+            if captured.get("schema_version") == "outcome_context_projection_v0":
+                source = captured.get("source_capture")
+                if not isinstance(source, dict):
+                    raise ValueError("scoped context has no original source reference")
+                _, _, source_payload = _read_capture(source.get("path"), source.get("sha256"))
+            selected = _turn_envelope_result("captured_outcome", payload=captured, outcome_ref=outcome_ref,
+                **({"source_payload": source_payload} if source_payload is not None else {}))
+            raw = (json.dumps(selected, ensure_ascii=False, separators=(",", ":")) + "\n").encode("utf-8")
+            selection_digest = hashlib.sha256(raw).hexdigest()
+            # File integrity alone cannot bind cursors in a derived rendering:
+            # the same file may be read by an upgraded projection owner. Never
+            # mix selected JSON layouts under an unchanged original-file SHA.
+            if expected_selection is not None and expected_selection != selection_digest:
+                raise ValueError("selected JSON digest mismatch; recover the original complete capture")
+            if offset > 0 and expected_selection is None:
+                raise ValueError("outcome continuation requires the returned selection-sha256; follow next_command")
         if offset > len(raw) or (offset < len(raw) and raw[offset] & 0xC0 == 0x80):
             raise ValueError("capture offset is outside the file or not a UTF-8 boundary; use the returned next offset")
         return _capture_page(path, raw, digest, offset=offset, size=size,
             output_limit=output_limit, instruction=bool(context_dir),
             presentation=presentation, next_page_bytes=next_page_bytes,
-            next_page_output_bytes=next_page_output_bytes)
-    except (OSError, ValueError, RecursionError) as exc:
+            next_page_output_bytes=next_page_output_bytes, outcome_ref=outcome_ref,
+            selection_sha256=selection_digest)
+    except (OSError, ValueError, RecursionError, EffectRuntimeRemoteError, EffectRuntimeStartupError) as exc:
         # Recovery is presentation-only: keep failures separate from quota's
         # collection error path and retain the existing receipt/capture.
         return {"ok": False, "schema_version": "context_capture_page_v1",
-            "reason": str(exc) if isinstance(exc, ValueError) else "original capture could not be read",
+            "reason": str(exc) if isinstance(exc, ValueError) else (
+                "captured outcome reader is unavailable; use the original full-capture reader"
+                if recovery_command else "original capture could not be read"),
+            **({"recovery_command": recovery_command} if recovery_command else {}),
             "recovery": "Restore the original digest-bound capture; no guard, admission or spend was performed."}
 
 
@@ -152,14 +194,22 @@ def native_context_capture_view(projected: dict[str, Any], directory: Path, *,
 
     if max_page_output_bytes is not None and max_page_output_bytes <= 0:
         raise QuotaDecisionTransportError("Native context output limit must be positive.")
-    wire = directory / "delivery.json"
-    write_decision_capture(wire, projected)
-    reference = {"path": str(wire), "sha256": hashlib.sha256(wire.read_bytes()).hexdigest()}
+    reference = None
     mode, target = "full", None
     if projected.get("schema_version") == "context_delivery_v1":
         mode, target = projected["mode"], projected["target"]
+        # Only the explicitly scoped artifact replaces the caller's original
+        # recovery route. Ordinary captures keep their exact historical bytes.
+        if projected["recovery"].get("presentation_schema") == "outcome_context_projection_v0":
+            metadata = {**(metadata or {}), "recovery": projected["recovery"]}
         if mode == "full":
             reference = {key: projected["recovery"][key] for key in ("path", "sha256")}
+    if reference is None:
+        # Full delivery already pages its immutable recovery. Do not serialize
+        # a duplicate wire file that no reader or lifecycle receipt consumes.
+        wire = directory / "delivery.json"
+        write_decision_capture(wire, projected)
+        reference = {"path": str(wire), "sha256": hashlib.sha256(wire.read_bytes()).hexdigest()}
     page_bytes, output_bytes = native_context_page_budget()
     if max_page_output_bytes is not None:
         output_bytes = min(output_bytes, max_page_output_bytes)
@@ -490,6 +540,15 @@ def handoff_decision(directory: Path | None, socket_path: str | None,
 def handle_native_context_command(args: argparse.Namespace, *, print_payload: Callable[..., None],
                                   registry_path: Path | None = None, runtime_root_arg: str | None = None) -> int | None:
     """Dispatch native presentation/lifecycle IO before any quota authority read."""
+    if getattr(args, "selection_sha256", None) is not None and (args.quota_command != "context-read"
+            or getattr(args, "outcome_ref", None) is None):
+        print_payload({"ok": False, "reason": "selection-sha256 requires context-read with outcome-ref"},
+            args.format, lambda value: str(value["reason"]))
+        return 1
+    if getattr(args, "outcome_ref", None) is not None and args.quota_command not in {"context-read", "replan-view"}:
+        print_payload({"ok": False, "reason": "outcome-ref is only supported by context-read and replan-view"},
+            args.format, lambda value: str(value["reason"]))
+        return 1
     has_output_limit = getattr(args, "max_page_output_bytes", None) is not None
     budget_scope_error = ("max-page-output-bytes is only supported by JSON context-read "
         "or enabled native should-run/context-project")
@@ -606,6 +665,8 @@ def read_replan_view(args: argparse.Namespace) -> dict[str, Any]:
 
     if not args.decision_file or not args.source_hash:
         raise ValueError("replan-view requires --decision-file and --source-hash")
+    if args.outcome_ref == "":
+        raise ValueError("replan-view outcome-ref requires an exact reference; use context-read for the captured index")
     path = Path(args.decision_file).expanduser()
     with path.open(encoding="utf-8") as stream:
         payload = json.load(stream)

loopx/cli_commands/quota_registration.py

@@ -43,7 +43,9 @@ def register_quota_command(
         help="Use status for all groups, plan for next-turn groups, should-run for one goal, context-read for bounded original-capture recovery without a guard, context-project to present an existing digest-bound quota capture with native full/delta and a full-read fallback, context-hook for opt-in native Codex lifecycle IO, context-config for native hook/recovery configuration without installation or trust changes, monitor-poll for no-spend quiet poll evidence, scheduler-ack for successful Codex App RRULE state, scheduler-fail-current to suppress a repeated failed host update pair, spend-slot for accounting, or void-slot for a non-destructive accounting correction.",
     )
     quota_parser.add_argument("--decision-file", help="Original captured decision for read-only replan-view.")
-    quota_parser.add_argument("--outcome-ref", help="Exact captured acceptance outcome reference for replan-view.")
+    quota_parser.add_argument("--outcome-ref", nargs="?", const="",
+        help="For context-read: omit the value to list captured outcomes, or supply an exact reference to page its full detail. "
+             "For replan-view: exact captured acceptance outcome reference. Neither changes required-read obligations.")
     quota_parser.add_argument("--source-hash", help="Original source_decision_hash for replan-view.")
     quota_parser.add_argument("--replan-stage", choices=["review", "writeback", "acceptance", "explore"], default="review",
                               help="Read decision context before opening its writeback format.")
@@ -225,6 +227,9 @@ def register_quota_command(
              "Use for shared Codex App hook configuration; cannot combine with --codex-context-dir. "
              "context-config renders native configuration for the exact Goal/Agent; no host configuration, trust, guard or spend are changed.")
     quota_parser.add_argument("--capture-sha256", help="For context-read/context-project: exact SHA-256 supplied with the capture reference.")
+    quota_parser.add_argument("--selection-sha256",
+        help="For context-read with outcome-ref: selected JSON byte digest returned by the first page. "
+             "Required on continuation; follow next_command rather than computing it.")
     quota_parser.add_argument("--offset", type=int, default=0, help="For context-read: byte offset returned by the previous page.")
     quota_parser.add_argument("--page-bytes", type=int,
         help="For context-read: maximum UTF-8 content bytes per page (JSON default: 64000 under a 64000-byte stdout budget; Markdown: 4096). "