Present generated Goal objectives once on native first reads

benchmark/runtime/RUNTIME.md

@@ -180,9 +180,10 @@ initial review and writeback routes for an admitted replan from the verified
 capture. Their readers retain the original guard's path and hash. The
 projection command also carries the Goal, Agent, Turn and private cache location. The
 native tool presents the saved decision using the same native lifecycle projector;
-no guard, checkpoint, admission or debit is repeated. Cold delivery supplies the
-full body and its target; qualified warm delivery supplies the exact delta.
-The small response points to bounded reads and an original full-read fallback.
+no guard, checkpoint, admission or debit is repeated. The file view can be full or a qualified exact delta. With `--initial-read`,
+the bounded response includes a first selection page and its continuation in
+top-level `content` and `next_command`. Forward them to the model along with
+`navigation`; use its original full-read fallback for literal recovery.
 Invalidated or unavailable baselines keep full input. If enabled native preparation cannot
 establish safe recovery, the command returns a typed transport failure with the
 original full-read reference and receipt; use that explicit reference without

benchmark/runtime/context_delivery.py

@@ -28,7 +28,9 @@ def preflight_context_instruction(env: dict[str, str], path: Path, digest: str,
         "--goal-id", env["LOOPX_GOAL_ID"], "--agent-id", env["LOOPX_AGENT_ID"],
         "--turn-instance-id", turn_id, "--initial-read", "--context-delivery", "on", "--runtime-profile", "generic_cli", "--codex-context-dir", env["LOOPX_CODEX_CONTEXT_DIR"]]
     recovery_index = [*prefix, "context-read", *source, "--list"]
-    return ("Present that same immutable packet with:\n```sh\n" + shlex.join(project)
+    return ("Run this command directly and deliver its complete bounded stdout to the model. "
+        "Keep top-level content and next_command when present; navigation alone omits the returned page. "
+        "Preserve any failure response.\n```sh\n" + shlex.join(project)
         + "\n```\n" + ("" if guide_in_host else CONTEXT_CAPTURE_NAVIGATION_GUIDANCE)
         + "If the projection command itself fails, recover necessary facts from the same original capture. "
         "When pointers are unknown, discover them with:\n```sh\n" + shlex.join(recovery_index)

docs/reference/required-work-context.md

@@ -110,6 +110,17 @@ current source witnesses and unknown fields remain literal. Cold delivery and
 full-capture recovery retain the complete original text, including arbitrary
 sections, comments and final stops.
 
+The native initial selection may render a generated Objective once when the
+frontmatter entry and the entire generated Markdown blockquote agree exactly
+under the writer's format. The selected text is marked as a presentation.
+Literal `--select` and full recovery on the same immutable capture retain the
+original bytes. Differing or unrecognized formats keep both values. The native
+read observer does not count the presented leaf as reading its literal source.
+
+原生首读仅在生成的 frontmatter 目标与正文整段生成引用相符时展示一次目标。
+首读会标记这一呈现；同一捕获的显式选读与完整恢复保留原字节。两处不一致
+或格式无法确认时保留原文。首读观察不会把呈现文本记为已读原始字段。
+
 Enabled native quota delivery separates known acceptance requirements from their
 supporting material before full/delta presentation. The TypeScript acceptance
 projection keeps every scoped outcome objective, requirement description,

loopx/cli_commands/quota_capture.py

@@ -19,7 +19,7 @@
 from ..control_plane.quota.error_codes import QuotaCommandValidationError, QuotaDecisionTransportError
 from ..control_plane.turn_driver.decision_capture import (
     context_index_fields as _context_index_fields,
-    context_selection_bytes as _context_selection_bytes,
+    context_selection_binding, write_initial_context_read_reference,
     context_selection_view as _context_selection_view,
     validate_handoff_socket, CONTEXT_CAPTURE_READ_GUIDANCE, CONTEXT_CAPTURE_PAGE_GUIDANCE, context_capture_read_command,
     CONTEXT_CAPTURE_NAVIGATION_GUIDANCE,
@@ -201,8 +201,12 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An
         if pinned_selection is not None:
             if selectors is not None or initial_read or list_fields or list_depth is not None or selection_digest is not None:
                 raise ValueError("a pinned selection reference cannot change selectors; use the original capture reference")
-            if (set(pinned_selection) != {"select", "initial_read", "list_fields", "list_depth", "sha256"}
+            if (set(pinned_selection) - {"present_goal_narrative"} !=
+                    {"select", "initial_read", "list_fields", "list_depth", "sha256"}
                     or type(pinned_selection["initial_read"]) is not bool or type(pinned_selection["list_fields"]) is not bool
+                    or ("present_goal_narrative" in pinned_selection and (
+                        pinned_selection["present_goal_narrative"] is not True
+                        or pinned_selection["list_fields"] or pinned_selection["initial_read"]))
                     or not isinstance(pinned_selection["sha256"], str)
                     or BARE_SHA256_PATTERN.fullmatch(pinned_selection["sha256"]) is None):
                 raise ValueError("reader selection reference is incomplete or unsupported")
@@ -247,7 +251,9 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An
             pinned_explicit = pinned_selection is not None and not initial_read
             plan, selected_raw, expected = _context_selection_view(payload, select=selectors,
                 initial_read=initial_read, list_fields=list_fields, list_depth=list_depth,
-                expected_sha256=selection_digest, preserve_selection_order=pinned_explicit)
+                expected_sha256=selection_digest, preserve_selection_order=pinned_explicit,
+                present_goal_narrative=(pinned_selection.get("present_goal_narrative") is True
+                    if pinned_selection is not None else None))
             if rejection := plan.get("rejected_selection"):
                 # TS alone resolves the nearest container. Reuse the original
                 # escaped tokens, without parsing diagnostic prose or traversing
@@ -273,9 +279,9 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An
                     # Pin the resolved addresses, not the initial navigation
                     # recipe. Existing explicit readers can replay these exact
                     # bytes after a navigation change, with both digests checked.
-                    reference_file = write_context_read_reference(path, digest, selection={
-                        "select": selectors, "initial_read": False,
-                        "list_fields": list_fields, "list_depth": list_depth, "sha256": expected})
+                    reference_file = write_context_read_reference(path, digest, selection=context_selection_binding(
+                        selectors, list_fields=list_fields, list_depth=list_depth, sha256=expected,
+                        presented=bool(plan.get("presentation_pointers"))))
                 except OSError:
                     # A verified read must not require a writable capture root.
                     # Legacy cursors pin the same capture, selection and digest.
@@ -546,11 +552,9 @@ def pin_initial_selection(plan: dict[str, Any]) -> None:
         nonlocal initial_read_pointers, initial_read_reference
         selections = plan["selections"]
         initial_read_pointers = [item["pointer"] for item in selections]
-        selected = _context_selection_bytes(current_document, selections, list_fields=False, list_depth=None)
         try:
-            initial_read_reference = write_context_read_reference(Path(current["path"]), current["sha256"], selection={
-                "select": initial_read_pointers, "initial_read": False, "list_fields": False, "list_depth": None,
-                "sha256": hashlib.sha256(selected).hexdigest()})
+            initial_read_reference = write_initial_context_read_reference(Path(current["path"]), current["sha256"],
+                current_document, selections)
         except OSError:
             # Read-only capture roots keep the existing initial recipe;
             # metadata is optional and never a read/authority receipt.

loopx/control_plane/goals/active_state_metadata.py

@@ -38,6 +38,34 @@ def markdown_blockquote(value: str) -> str:
     return "\n".join(f"> {escape(line, quote=False)}" for line in value.splitlines())
 
 
+def goal_narrative_without_objective_mirror(state_text: str) -> str | None:
+    """Render the generated Goal template once; leave literal/legacy text alone.
+
+    Require the writer's header, canonical JSON string and entire generated
+    blockquote. A second metadata entry, customized heading or conflicting
+    objective is user data and does not qualify.
+    """
+    if not state_text.startswith("---\n"):
+        return None
+    header, boundary, body = state_text[4:].partition("\n---\n")
+    if not boundary:
+        return None
+    rows = header.split("\n")
+    keys = [i for i, line in enumerate(rows) if ":" in line and line.split(":", 1)[0].strip() == "objective"]
+    if len(keys) != 1:
+        return None
+    metadata, original_body = split_state_frontmatter(state_text)
+    objective = metadata.get("objective")
+    if (original_body != body or not objective
+            or "\n".join(objective.splitlines()) != objective.removesuffix("\n")
+            or rows[keys[0]] != f"objective: {markdown_frontmatter_string(objective)}"):
+        return None
+    prefix = "\n# Active Goal State\n\n## Objective\n\n"
+    if not body.startswith(prefix + markdown_blockquote(objective) + "\n\n## "):
+        return None
+    del rows[keys[0]]
+    return "---\n" + "\n".join(rows) + boundary + body
+
 def active_state_section_text(state_text: str, heading: str) -> str:
     _, body = split_state_frontmatter(state_text)
     sections = active_state_sections(

loopx/control_plane/turn_driver/decision_capture.py

@@ -559,26 +559,81 @@ def context_selection_transport(payload: dict[str, Any], select: list[str] | Non
     return result
 
 
-def context_selection_bytes(payload: dict[str, Any], selections: list[dict[str, Any]],
-                             *, list_fields: bool, list_depth: int | None) -> bytes:
-    """Render TS-verified addresses from original, digest-checked JSON values."""
+_GOAL_SOURCE_PREFIX = ["interaction_contract", "agent_channel", "work_context", "sources"]
+_GOAL_NARRATIVE_PRESENTATION = (
+    "The generated frontmatter objective appears once under ## Objective. "
+    "Use --select on the same capture at this pointer for verbatim source text."
+)
+
+
+def _first_read_goal_text(payload: dict[str, Any], path: list[Any], value: Any) -> str | None:
+    # Only render the typed selector's registered Goal text leaf. The source,
+    # enclosing containers, exact detail reads and unknown shapes stay literal.
+    if (not isinstance(value, str) or path[:4] != _GOAL_SOURCE_PREFIX
+            or len(path) != 7 or type(path[4]) is not int or path[5:] != ["content", "text"]):
+        return None
+    interaction = payload["interaction_contract"]
+    source = interaction["agent_channel"]["work_context"]["sources"][path[4]]
+    content = source["content"]
+    if (interaction.get("schema_version") != "loopx_interaction_contract_v0"
+            or source.get("source") != "goal_state" or content.get("ok") is False
+            or content.get("read_complete") is False or not content.get("state_file")
+            or not (content.get("full_source") is True or content.get("excluded_todo_regions"))):
+        return None
+    from ..goals.active_state_metadata import goal_narrative_without_objective_mirror
+
+    text = goal_narrative_without_objective_mirror(value)
+    # Allow room for the presentation notice and immutable selection metadata.
+    return text if text is not None and len(value) - len(text) > len(_GOAL_NARRATIVE_PRESENTATION) + 96 else None
+
+
+def _context_selection_bytes(payload: dict[str, Any], selections: list[dict[str, Any]],
+                             *, list_fields: bool, list_depth: int | None,
+                             present_goal_narrative: bool) -> tuple[bytes, list[str]]:
+    """Render TS-verified addresses from the original, digest-checked values."""
     views: list[dict[str, Any]] = []
+    presented: list[str] = []
     for item in selections:
         selected: Any = payload
-        for key in item["path"]:
+        path, pointer = item["path"], item["pointer"]
+        for key in path:
             selected = selected[key]
         if list_fields:
-            views.append({"pointer": item["pointer"], "fields": context_index_fields(payload, item["children"])})
+            views.append({"pointer": pointer, "fields": context_index_fields(payload, item["children"])})
         else:
-            views.append({"pointer": item["pointer"], "value": selected})
+            text = _first_read_goal_text(payload, path, selected) if present_goal_narrative else None
+            if text is None:
+                views.append({"pointer": pointer, "value": selected})
+            else:
+                presented.append(pointer)
+                views.append({"pointer": pointer, "value": text, "presentation": _GOAL_NARRATIVE_PRESENTATION})
     view_options = {"list_depth": list_depth} if list_depth is not None and list_depth != 1 else {}
-    return json.dumps({"schema_version": "context_capture_index_v1" if list_fields else "context_capture_selection_v1",
+    raw = json.dumps({"schema_version": "context_capture_index_v1" if list_fields else "context_capture_selection_v1",
         "selections": views, **view_options}, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") + b"\n"
+    return raw, presented
+
+
+def context_selection_binding(select: list[str], *, list_fields: bool, list_depth: int | None,
+                              sha256: str, presented: bool = False) -> dict[str, Any]:
+    return {"select": select, "initial_read": False, "list_fields": list_fields, "list_depth": list_depth,
+        "sha256": sha256, **({"present_goal_narrative": True} if presented else {})}
+
+
+def write_initial_context_read_reference(path: Path, digest: str, payload: dict[str, Any],
+                                        selections: list[dict[str, Any]], *,
+                                        observed_read: dict[str, Any] | None = None) -> Path:
+    raw, presented = _context_selection_bytes(payload, selections, list_fields=False, list_depth=None,
+        present_goal_narrative=True)
+    return write_context_read_reference(path, digest,
+        selection=context_selection_binding([item["pointer"] for item in selections], list_fields=False,
+            list_depth=None, sha256=hashlib.sha256(raw).hexdigest(), presented=bool(presented)),
+        observed_read=observed_read)
 
 
 def context_selection_view(payload: dict[str, Any], *, select: list[str] | None,
                            initial_read: bool, list_fields: bool, list_depth: int | None,
                            expected_sha256: str | None, preserve_selection_order: bool = False,
+                           present_goal_narrative: bool | None = None,
                            ) -> tuple[dict[str, Any], bytes | None, str | None]:
     """Replay the TS selector and verify the original digest-pinned view."""
     from ..quota.turn_envelope import _turn_envelope_result
@@ -592,15 +647,24 @@ def context_selection_view(payload: dict[str, Any], *, select: list[str] | None,
         recover_selection_errors=not initial_read, **selection, **depth, **order)
     if plan.get("rejected_selection"):
         return plan, None, None
-    raw = context_selection_bytes(payload, plan["selections"], list_fields=list_fields, list_depth=list_depth)
-    digest = hashlib.sha256(raw).hexdigest()
+    presentation = initial_read if present_goal_narrative is None else present_goal_narrative
+    def render(plan: dict[str, Any], present: bool):
+        raw, displayed = _context_selection_bytes(payload, plan["selections"], list_fields=list_fields,
+            list_depth=list_depth, present_goal_narrative=present)
+        return raw, hashlib.sha256(raw).hexdigest(), displayed
+
+    raw, digest, displayed = render(plan, presentation)
     if preserve_selection_order and expected_sha256 != digest:
         # Historical explicit references may pin canonical output despite
         # retaining caller order. Only the same source and addresses can replay.
         plan = _turn_envelope_result("context_select", context=context,
             list_fields=list_fields, **selection, **depth)
-        raw = context_selection_bytes(payload, plan["selections"], list_fields=list_fields, list_depth=list_depth)
-        digest = hashlib.sha256(raw).hexdigest()
+        raw, digest, displayed = render(plan, presentation)
+    if (expected_sha256 is not None and expected_sha256 != digest
+            and initial_read and present_goal_narrative is None and displayed):
+        # An unpinned initial cursor may have started before presentation was
+        # enabled. Its original digest must still match the literal view.
+        raw, digest, displayed = render(plan, False)
     if expected_sha256 is not None and expected_sha256 != digest:
         raise ValueError("selected view digest changed; restore the original selection and reader")
-    return plan, raw, digest
+    return {**plan, "presentation_pointers": displayed} if displayed else plan, raw, digest

loopx/control_plane/turn_driver/native_context_selection.py

@@ -59,15 +59,18 @@ def verified_context_selection(state: dict[str, Any], page: dict[str, Any]) -> d
         return None
     _, raw, source = read_context_capture_file(str(physical), source_digest, private=True)
     actual_digest = source_digest
+    presentation_pointers: set[str] = set()
     if selected is not None or "selected_pointers" in page:
         selection, selected_raw, actual_digest = context_selection_view(source, select=pointers,
             initial_read=initial_read, list_fields=False,
             list_depth=selected.get("list_depth") if selected else None,
-            expected_sha256=view_digest, preserve_selection_order=selected is not None and not initial_read)
+            expected_sha256=view_digest, preserve_selection_order=selected is not None and not initial_read,
+            present_goal_narrative=bool(selected and selected.get("present_goal_narrative") is True))
         if selected_raw is None:
             return None
         raw = selected_raw
         pointers = [item["pointer"] for item in selection["selections"]]
+        presentation_pointers = set(selection.get("presentation_pointers", []))
     if actual_digest != view_digest or len(raw) != total or raw[offset:offset + len(actual)] != actual:
         return None
     end = offset + len(actual)
@@ -83,7 +86,9 @@ def verified_context_selection(state: dict[str, Any], page: dict[str, Any]) -> d
         # Native hook processes need not inherit the shell's CODEX_THREAD_ID.
         if observation_current(observation, physical, source_digest, hook_session=state.get("session_id")):
             inherited = observation["fields"]
-    sections = _turn_envelope_result("context_read_sections", context=source, select=pointers,
-        consumed_pointers=inherited)["sections"]
+    # A display edit is not an observation of its literal source text.
+    literal_pointers = [pointer for pointer in pointers if pointer not in presentation_pointers]
+    sections = _turn_envelope_result("context_read_sections", context=source, select=literal_pointers,
+        consumed_pointers=inherited)["sections"] if literal_pointers else {}
     return {**page, "schema_version": "context_capture_page_v1", "sha256": view_digest,
         "selected_pointers": pointers, "source_sha256": current["sha256"], "read_sections": sections}