diff --git a/docs/architecture/rfcs/target-yardstick-stride-decisions-v0.md b/docs/architecture/rfcs/target-yardstick-stride-decisions-v0.md
index 0aeffb0e98..8b4adc4cfa 100644
--- a/docs/architecture/rfcs/target-yardstick-stride-decisions-v0.md
+++ b/docs/architecture/rfcs/target-yardstick-stride-decisions-v0.md
@@ -535,6 +535,15 @@ context projects only that Todo's current target, a short next-choice cue and
 counterevidence references. Its existing question-scoped read command exposes
 the full decision, exact-profile feedback and prior execution attribution; the
 default projection does not repeat that full record.
+Before a current work decision exists, the same opt-in starting read can expose
+bounded measurement feedback from the Todo's canonical question links. This
+reuses Explore's exact-profile comparison owner, includes source references and
+omitted-group navigation, and does not infer a target, accepted outcome metric
+or proxy classification. A shared question link is evidence association, not
+immutable attribution of execution to that Todo. Recording the first decision
+remains optional; default-off and exact-field first reads retain their prior
+semantics. Production File/SQLite reader qualification and actual model benefit
+are separate obligations.
 Counterevidence requests reassessment at the next concrete choice; it is not a
 new hard gate, fidelity certification, permission to waive correctness or an
 instruction to replan every Turn. At a later meaningful capture, existing
diff --git a/docs/architecture/rfcs/target-yardstick-stride-decisions-v0.zh-CN.md b/docs/architecture/rfcs/target-yardstick-stride-decisions-v0.zh-CN.md
index 9aee8eeefa..d19f45ab96 100644
--- a/docs/architecture/rfcs/target-yardstick-stride-decisions-v0.zh-CN.md
+++ b/docs/architecture/rfcs/target-yardstick-stride-decisions-v0.zh-CN.md
@@ -415,6 +415,11 @@ Explore evidence 可用；开启该设置不启动工作或自动开启其他 ca
 普通 task-scoped Explore context 只投影该 Todo 当前目标、下一选择短提示和反证引用。
 既有 question-scoped 扩读命令返回完整决策、精确 profile 的反馈和前次执行归属；
 默认投影不重复完整记录。
+尚无当前工作决策时，同一 opt-in 首读可呈现来自该 Todo 规范问题关联的有界测量反馈。
+复用 Explore 精确 profile 比较 owner，保留来源引用及被省略组的读取入口，不推断目标、
+验收成果指标或代理分类。共享问题关联只表示证据关联，不是该 Todo 执行测量的不可变归属。
+首次决策仍为可选记录；默认关闭及按字段首次读取保留原语义。真实 File/SQLite reader
+资格与实际模型收益分别验收。
 反证要求在下一个具体选择处重新判断，不构成新的硬门、fidelity 认证、放松正确性约束的
 权限或每轮重规划指令。后续有意义的 capture 可复用 execution observation，把前次
 提议方法连到实际方法、候选和证据，也允许有据采用不同方法。
diff --git a/loopx/capabilities/explore/README.md b/loopx/capabilities/explore/README.md
index daa00573c3..877a59c22a 100644
--- a/loopx/capabilities/explore/README.md
+++ b/loopx/capabilities/explore/README.md
@@ -139,6 +139,16 @@ The comparison uses findings from the current work, target, and acceptance
 reference. A different target or acceptance reference needs its own eligible
 pair, even when the measured metric, evaluator and inputs stay the same.
 The comparison is advisory and carries no retention or execution obligation.
+On an admitted work turn, the progressive starting read includes the enabled
+Yardstick's decision guidance and any compact cue for the selected Todo. Use
+the Explore context reader for the attachment template and scoped evidence.
+Before a first decision is recorded, the same opt-in entry shows up to three
+exact-profile measurement groups from questions canonically linked to that
+Todo, with omitted counts and the existing task reader. Distinct questions get
+a place before additional metrics from one question. These are declared
+comparisons, including shared findings; they do not establish which Todo ran a
+measurement or select an outcome/proxy profile. Default-off input and the
+benchmark's optional exact-field first read retain their existing behavior.
 
 Manually captured measured findings also appear in the existing result pages
 and, when linked to eligible work, its exact-source comparison cards. They need
diff --git a/loopx/capabilities/explore/turn_context.py b/loopx/capabilities/explore/turn_context.py
index 46af6028e3..6413f2ed91 100644
--- a/loopx/capabilities/explore/turn_context.py
+++ b/loopx/capabilities/explore/turn_context.py
@@ -13,6 +13,7 @@
     dispatch_interaction_projection_hooks,
     TURN_START_HOOK_RESULT_SCHEMA_VERSION,
     TurnStartHookRegistration,
+    TurnStartContextProjector,
     dispatch_turn_start_hooks,
 )
 from ...control_plane.effect_runtime import effect_runtime_request_scope, effect_runtime_result
@@ -155,6 +156,30 @@ def explore_turn_context(
     ))
 
 
+def turn_context_projector(
+    *, registry_path: Path, runtime_root: Path, goal_id: str,
+    agent_id: str | None, todo_id: str | None = None,
+) -> TurnStartContextProjector | None:
+    """Bind an entrypoint's existing read request after governing selection.
+
+    Merely enabling a Goal or supplying unrelated hooks cannot request this
+    read. The decision owner supplies facts; this adapter changes no admission.
+    """
+    if not agent_id:
+        return None
+
+    def project(payload: Mapping[str, Any], dispatch: Mapping[str, Any] | None) -> dict[str, Any] | None:
+        channel = (payload.get("interaction_contract") or {}).get("agent_channel") or {}
+        selected = payload.get("selected_todo") or {}
+        focus = (selected.get("todo_id") if payload.get("normal_delivery_allowed") is True
+                 and channel.get("delivery_allowed") is True and selected.get("status") == "open" else None)
+        return extend_turn_start_dispatch(dict(dispatch or {}),
+            registry_path=registry_path, runtime_root=runtime_root,
+            goal_id=goal_id, agent_id=agent_id, todo_id=todo_id, selected_todo_id=focus)
+
+    return project
+
+
 def extend_turn_start_dispatch(
     dispatch: dict[str, Any] | None,
     *,
@@ -163,12 +188,20 @@ def extend_turn_start_dispatch(
     goal_id: str,
     agent_id: str | None,
     todo_id: str | None = None,
+    selected_todo_id: str | None = None,
 ) -> dict[str, Any] | None:
     if not agent_id:
         return dispatch
-    _, graph, gate = _policy(registry_path, goal_id, runtime_root)
+    goal, graph, gate = _policy(registry_path, goal_id, runtime_root)
     if not graph and not gate["enabled"]:
         return dispatch
+    if selected_todo_id is not None:
+        from ..todo_replan_cadence.machine_defaults import work_yardstick_enabled
+
+        # This optional policy's current-work card follows the resolved work
+        # identity. With the policy off, preserve the caller's original scope.
+        if work_yardstick_enabled(goal, runtime_root):
+            todo_id = selected_todo_id
 
     def produce() -> dict[str, Any]:
         return {
diff --git a/loopx/cli_commands/quota.py b/loopx/cli_commands/quota.py
index 741e45b729..22ad811942 100644
--- a/loopx/cli_commands/quota.py
+++ b/loopx/cli_commands/quota.py
@@ -258,10 +258,6 @@ def _dispatch_quota_turn_start_hooks(
         )
         dispatch = extend_cadence_turn_start_dispatch(dispatch, registry_path=registry_path,
             runtime_root=root, goal_id=args.goal_id, agent_id=args.agent_id)
-    if args.agent_id:
-        from ..capabilities.explore.turn_context import extend_turn_start_dispatch as extend_explore
-        dispatch = extend_explore(dispatch, registry_path=registry_path, runtime_root=root,
-            goal_id=args.goal_id, agent_id=args.agent_id, todo_id=args.todo_id)
     local_private_state_mutated = bool(dispatch.get("uncertain_local_effects")) or any(
         isinstance(result, Mapping)
         and result.get("local_private_state_mutated") is not False
@@ -299,14 +295,12 @@ def _prepare_quota_command_execution(
     return goal_ref, turn_start_hook_dispatch, context
 
 
-def _attach_turn_start_hook_dispatch(
-    payload: dict[str, object],
-    dispatch: Mapping[str, object] | None,
-) -> None:
-    if dispatch and (dispatch.get("registered_count") or dispatch.get("failures")):
-        payload["turn_start_capability_hook_dispatch"] = {
-            key: value for key, value in dispatch.items() if key != "contexts"
-        }
+def _quota_turn_context_projector(args, *, registry_path: Path, runtime_root: Path):
+    from ..capabilities.explore.turn_context import turn_context_projector
+
+    return turn_context_projector(registry_path=registry_path, runtime_root=runtime_root,
+        goal_id=args.goal_id, agent_id=args.agent_id, todo_id=args.todo_id)
+
 
 def _emit_bounded_settlement(payload, args, *, runtime_root, reader_argv, print_payload):
     """Preserve the committed receipt even when bounded result delivery fails."""
@@ -603,6 +597,8 @@ def handle_quota_command(
                 retained_todo_id=retained_action_selection_todo_id,
             )
             heartbeat_receipt_existing = action_selection.receipt
+            context_projector = _quota_turn_context_projector(args,
+                registry_path=registry_path, runtime_root=runtime_root)
             payload = _export("build_live_quota_should_run_decision")(
                 status_payload,
                 goal_id=args.goal_id,
@@ -636,13 +632,13 @@ def handle_quota_command(
                 turn_instance_id=heartbeat_turn_id,
                 interaction_projection_hooks=interaction_projection_hooks,
                 turn_start_hook_dispatch=turn_start_hook_dispatch,
+                turn_context_projector=context_projector,
                 context_navigation=(
                     native_context_activation(args)["progressive_enabled"]
                     and not bool(getattr(args, "turn_envelope", False))
                 ),
                 goal_ref=goal_ref,
             )
-            _attach_turn_start_hook_dispatch(payload, turn_start_hook_dispatch)
             action_selection_preflight = (
                 _export("reconcile_requested_quota_action_selection")(
                     payload,
@@ -713,6 +709,7 @@ def handle_quota_command(
                         context=context,
                         interaction_projection_hooks=interaction_projection_hooks,
                         turn_start_hook_dispatch=turn_start_hook_dispatch,
+                        turn_context_projector=context_projector,
                         action_selection=action_selection,
                         cache_metadata=cache_metadata,
                         goal_ref=goal_ref,
diff --git a/loopx/cli_commands/quota_monitor_poll.py b/loopx/cli_commands/quota_monitor_poll.py
index 501f302def..97620c099e 100644
--- a/loopx/cli_commands/quota_monitor_poll.py
+++ b/loopx/cli_commands/quota_monitor_poll.py
@@ -6,7 +6,7 @@
 from typing import TYPE_CHECKING
 
 from ..control_plane.coordination.local_authority import CanonicalAuthorityRequired
-from ..control_plane.capability_hooks import InteractionProjectionHookRegistration
+from ..control_plane.capability_hooks import InteractionProjectionHookRegistration, TurnStartContextProjector
 from ..control_plane.quota.effective_action import EffectiveAction
 from ..control_plane.quota.error_codes import HeartbeatReceiptIdentityConflictError
 from ..control_plane.quota.heartbeat_receipt import (
@@ -186,6 +186,7 @@ def record_automatic_heartbeat_stall(
     status_collector: Callable[..., dict[str, object]],
     decision_builder: Callable[..., dict[str, object]],
     host_observation_resolver: Callable[..., object],
+    turn_context_projector: TurnStartContextProjector | None = None,
 ) -> tuple[dict[str, object], dict[str, object], object, str]:
     """Commit and reproject the automatic no-spend heartbeat observation."""
 
@@ -263,6 +264,7 @@ def record_automatic_heartbeat_stall(
         turn_instance_id=turn_id,
         interaction_projection_hooks=interaction_projection_hooks,
         turn_start_hook_dispatch=turn_start_hook_dispatch,
+        turn_context_projector=turn_context_projector,
         goal_ref=goal_ref,
     )
     rebuilt["heartbeat_stall_writeback"] = {
diff --git a/loopx/cli_commands/turn.py b/loopx/cli_commands/turn.py
index 7a4ac7350e..2601fa604e 100644
--- a/loopx/cli_commands/turn.py
+++ b/loopx/cli_commands/turn.py
@@ -225,10 +225,6 @@ def handle_turn_command(
             from ..capabilities.semantic_preference.agent_preferences import extend_turn_start_dispatch as extend_preferences
             turn_start_hook_dispatch = extend_preferences(turn_start_hook_dispatch, runtime_root=runtime_root,
                 registry_path=registry_path, goal_id=args.goal_id, agent_id=args.agent_id)
-            from ..capabilities.explore.turn_context import extend_turn_start_dispatch as extend_explore
-            turn_start_hook_dispatch = extend_explore(turn_start_hook_dispatch,
-                registry_path=registry_path, runtime_root=runtime_root,
-                goal_id=args.goal_id, agent_id=args.agent_id, todo_id=getattr(args, "todo_id", None))
             if delegated_context is not None:
                 # Invocation material shares normal capture/delivery. It is not a
                 # capability observation, another decision or a new read command.
@@ -239,12 +235,17 @@ def handle_turn_command(
             # from the same live status, scheduler context and capability hooks, so
             # this Turn takes all of them -- and its later settle-against inputs --
             # from the shared decision owner instead of rebuilding them per command.
+            from ..capabilities.explore.turn_context import turn_context_projector
+
             decision_owner = build_fresh_turn_decision_owner(
                 args,
                 registry_path=registry_path,
                 runtime_root=runtime_root,
                 runtime_root_arg=runtime_root_arg,
                 turn_start_hook_dispatch=turn_start_hook_dispatch,
+                turn_context_projector=turn_context_projector(registry_path=registry_path,
+                    runtime_root=runtime_root, goal_id=args.goal_id, agent_id=args.agent_id,
+                    todo_id=getattr(args, "todo_id", None)),
                 goal_ref=goal_ref,
                 request_subject=delegated_context.get("request_subject") if delegated_context else None,
             )
diff --git a/loopx/cli_commands/turn_decision.py b/loopx/cli_commands/turn_decision.py
index 0bbdaf7ac3..9813ff5693 100644
--- a/loopx/cli_commands/turn_decision.py
+++ b/loopx/cli_commands/turn_decision.py
@@ -31,6 +31,7 @@
     periodic_report_pending_intent_interaction_hook,
 )
 from ..control_plane.quota.live_decision import build_live_quota_should_run_decision
+from ..control_plane.capability_hooks import TurnStartContextProjector
 from ..control_plane.quota.effective_action import EffectiveAction
 from ..control_plane.quota.turn_envelope import (
     TurnControllerInput, build_turn_controller_input, build_turn_envelope,
@@ -139,6 +140,7 @@ def _build_turn_decision(
     scheduler_execution_context: Mapping[str, Any] | SchedulerExecutionContextResolution,
     operator_inbox_urgency_projector: Callable[..., dict[str, Any]],
     turn_start_hook_dispatch: Mapping[str, Any] | None = None,
+    turn_context_projector: TurnStartContextProjector | None = None,
     goal_ref: Mapping[str, object] | None = None,
     request_subject: Mapping[str, Any] | None = None,
 ) -> Callable[..., dict[str, Any]]:
@@ -191,6 +193,7 @@ def build_turn_decision(
             receipt_bound_replan_obligation_id=bound_replan,
             turn_instance_id=turn_instance_id if bound_replan else None,
             turn_start_hook_dispatch=dict(turn_start_hook_dispatch or {}),
+            turn_context_projector=turn_context_projector,
             workspace_path=workspace_path,
             goal_ref=goal_ref,
             interaction_projection_hooks=(
@@ -304,6 +307,7 @@ def build_fresh_turn_decision_owner(
     runtime_root: Path,
     runtime_root_arg: str | None,
     turn_start_hook_dispatch: Mapping[str, Any] | None = None,
+    turn_context_projector: TurnStartContextProjector | None = None,
     goal_ref: Mapping[str, object] | None = None,
     request_subject: Mapping[str, Any] | None = None,
 ) -> FreshTurnDecisionOwner:
@@ -342,6 +346,7 @@ def build_fresh_turn_decision_owner(
             scheduler_execution_context=scheduler_execution_context,
             operator_inbox_urgency_projector=operator_inbox_urgency_projector,
             turn_start_hook_dispatch=turn_start_hook_dispatch,
+            turn_context_projector=turn_context_projector,
             goal_ref=goal_ref, request_subject=request_subject,
         ),
         requested_todo_id=getattr(args, "todo_id", None), request_subject=request_subject,
diff --git a/loopx/control_plane/capabilities/explore_turn_context.ts b/loopx/control_plane/capabilities/explore_turn_context.ts
index 969da23a9c..fb6556eb5e 100644
--- a/loopx/control_plane/capabilities/explore_turn_context.ts
+++ b/loopx/control_plane/capabilities/explore_turn_context.ts
@@ -18,7 +18,7 @@ import {projectExploreMeasurementFeedback} from "./explore_measurements.ts";
  * provenance and applicability; no timestamp or revision is normalized away. */
 export function exploreReusableContextPaths(content: JsonObject = {}): (string | number)[][] {
   return [
-    ...["recent_nodes", "recent_findings", "writeback_results", "measurement_feedback", "work_yardsticks", "work_yardstick_capture",
+    ...["recent_nodes", "recent_findings", "writeback_results", "measurement_feedback", "work_measurements", "work_yardsticks", "work_yardstick_capture",
       "path_delta_attachment_template", "linked_question_attachment_template",
       "optional_question_evolution_example", "optional_execution_observation_example", "result_writeback_guidance",
       "result_attachment_template", "optional_measurement_example", "measurement_guidance",
@@ -338,14 +338,34 @@ export function projectExploreTurnContext(params: JsonObject): JsonObject {
     ...(resultNode == null ? [] : ["--result-node", resultNode]),
   ) : null;
   const writeback = exploreResultWritebackAffordance();
-  const workCues = params.yardstick_enabled === true ? rows(params.work_todos)
+  const workTodos = params.yardstick_enabled === true ? rows(params.work_todos)
     .filter(todo => todo.role === "agent" && !todo.done && todo.claimed_by === agent
       && ["open", "pending", "in_progress", "blocked"].includes(String(todo.status))
-      && (todoId === null || todo.todo_id === todoId))
-    .map(todo => projectWorkYardstickCue({enabled: true, goal_id: goal, agent_id: agent, todo_id: todo.todo_id,
+      && (todoId === null || todo.todo_id === todoId)) : [];
+  const currentWorkCues = workTodos.map(todo => projectWorkYardstickCue({enabled: true, goal_id: goal, agent_id: agent, todo_id: todo.todo_id,
       findings: findings.filter(row => (todo.explore_result_node_refs as unknown[] ?? []).includes(row.node_id)),
       measurement_feedback: measurementFeedback}))
-    .filter((cue): cue is JsonObject => cue !== null && (resultNode === null || cue.node_id === resultNode)) : [];
+    .filter((cue): cue is JsonObject => cue !== null);
+  const workCues = currentWorkCues.filter(cue => resultNode === null || cue.node_id === resultNode);
+  // A first decision may use measurements recorded before any Yardstick
+  // declaration. Reuse exact-profile feedback through canonical question links;
+  // those links do not establish which Todo executed a shared finding.
+  const workMeasurements = workTodos.flatMap(todo => {
+    if (todo.status !== "open" || currentWorkCues.some(cue => (cue.binding as JsonObject).todo_id === todo.todo_id)) return [];
+    const linked = new Set(requireStringArray(todo.explore_result_node_refs ?? [], "Explore task question refs"));
+    const available = measurementFeedback.filter(row => linked.has(row.node_id));
+    if (!available.length) return [];
+    // Give distinct linked questions a place before filling the remaining
+    // budget with further metrics. No metric is inferred to be the outcome.
+    const first = new Map<string, typeof available[number]>();
+    for (const row of available) if (!first.has(row.node_id)) first.set(row.node_id, row);
+    const representatives = new Set(first.values());
+    const visible = [...representatives, ...available.filter(row => !representatives.has(row))].slice(0, 3);
+    return [{todo_id: todo.todo_id, measurement_feedback: visible,
+      omitted_measurement_feedback: available.length - visible.length,
+      details_command: command("turn-context", "--agent-id", agent, "--todo-id", String(todo.todo_id), "--result-limit", "20"),
+      guidance: "Recorded comparisons for questions linked by this Todo; shared findings do not prove this Todo executed them. Match each metric and recorded scope to the accepted outcome before choosing work. No outcome/proxy profile has been selected here. Latest recorded is not current or retained, and best is not acceptance or permission. Use the existing task reader for omitted groups; a decision attachment remains optional."}];
+  });
   return {
     ok: true, goal_id: goal, agent_id: agent, graph_enabled: graph, harness_enabled: harness,
     ...(todoId === null ? {} : {todo_id: todoId}),
@@ -354,11 +374,13 @@ export function projectExploreTurnContext(params: JsonObject): JsonObject {
       counts: projection.counts ?? {},
       ...(params.yardstick_enabled === true ? {
         work_yardstick_capture: {optional: true, field: "work_yardstick_decision", schema_version: WORK_YARDSTICK_DECISION_SCHEMA,
-          guidance: "At a meaningful target, yardstick or next-method change, optionally attach the current decision to the existing Explore result. No change needs no attachment. Cite prior_finding_id from this Todo's current decision (null initially). Distinguish the accepted outcome's evaluation profile from an unqualified proxy; record counterevidence and bounded next work without changing user constraints. Read the ordinary task-scoped context for omitted decisions.",
+          guidance: "Before a proxy would discard a viable candidate or redirect work, check agreement with the accepted outcome in this scope. When agreement is unknown or the signals disagree, use one bounded accepted-outcome comparison if available within the existing constraints. At a meaningful target, yardstick or method change, optionally record that decision and its counterevidence in the same Todo's Explore result. Cite its prior_finding_id (null initially); omit the attachment when the decision is unchanged. Read the task-scoped Explore context for the template and omitted decisions.",
           template: {schema_version: WORK_YARDSTICK_DECISION_SCHEMA,
             target: {summary: "", scope: "", acceptance_ref: ""},
             authoritative: {metric: "", unit: "", higher_is_better: true, comparison_scope: "", input_revision: "", evaluation_ref: ""},
             proxy: null, counterevidence_refs: [], next_action: {method_id: "", reason: "", cost_bound: ""}, prior_finding_id: null}},
+        ...(workMeasurements.length ? {work_measurements: workMeasurements.slice(0, 3),
+          omitted_work_measurements: Math.max(0, workMeasurements.length - 3)} : {}),
         ...(workCues.length ? {work_yardsticks: workCues.slice(0, 3).map(cue => {
           if (resultNode !== null) return cue;
           const comparison = cue.search_comparison as JsonObject | undefined;
diff --git a/loopx/control_plane/capability_hooks.py b/loopx/control_plane/capability_hooks.py
index 7b1c482d7f..464830766f 100644
--- a/loopx/control_plane/capability_hooks.py
+++ b/loopx/control_plane/capability_hooks.py
@@ -69,6 +69,11 @@
 
 InteractionProjectionProducer = Callable[[], Mapping[str, Any]]
 TurnStartProducer = Callable[[], Mapping[str, Any]]
+# Invocation-owned read composition after selection; no persisted policy or
+# provider authority is created by supplying this adapter.
+TurnStartContextProjector = Callable[
+    [Mapping[str, Any], Mapping[str, Any] | None], dict[str, Any] | None
+]
 PostWritebackProducer = Callable[[Mapping[str, Any]], Mapping[str, Any]]
 
 
diff --git a/loopx/control_plane/quota/live_decision.py b/loopx/control_plane/quota/live_decision.py
index 23a98ebfab..0b94936bd1 100644
--- a/loopx/control_plane/quota/live_decision.py
+++ b/loopx/control_plane/quota/live_decision.py
@@ -13,6 +13,7 @@
 from ..goals.goal_frontier.fallback_source import live_fallback_authority_items
 from ..capability_hooks import (
     InteractionProjectionHookRegistration,
+    TurnStartContextProjector,
     dispatch_interaction_projection_hooks,
 )
 from .effect_program import (
@@ -220,8 +221,12 @@ def _project_turn_start_required_reads(
     turn_instance_id: str | None,
     registry_path: Path,
     runtime_root: Path,
-) -> bool:
-    """Order evidence before work and report whether the decision changed."""
+    context_projector: TurnStartContextProjector | None = None,
+) -> Mapping[str, Any] | None:
+    """Order evidence after selection and retain its source dispatch."""
+
+    if context_projector is not None:
+        dispatch = context_projector(payload, dispatch)
 
     # Keep failure observations even when no evidence read was produced. The
     # typed envelope projects their cache/dependent-action policy for the host.
@@ -231,7 +236,7 @@ def _project_turn_start_required_reads(
         }
     projected = _turn_start_required_reads(dispatch)
     if not projected:
-        return False
+        return dispatch
     existing = payload.get("required_reads")
     required_reads = (
         [dict(item) for item in existing if isinstance(item, Mapping)]
@@ -270,7 +275,7 @@ def _project_turn_start_required_reads(
         registry_path=str(registry_path),
         runtime_root=str(runtime_root),
     )
-    return True
+    return dispatch
 
 
 def _fresh_operator_inbox_observation_count(
@@ -595,6 +600,7 @@ def build_live_quota_should_run_decision(
     interaction_projection_hooks: Sequence[InteractionProjectionHookRegistration]
     | None = None,
     turn_start_hook_dispatch: Mapping[str, Any] | None = None,
+    turn_context_projector: TurnStartContextProjector | None = None,
     context_navigation: bool = False,
     workspace_path: Path | None = None,
     goal_ref: Mapping[str, object] | None = None,
@@ -756,7 +762,7 @@ def build_live_quota_should_run_decision(
             turn_start_hook_dispatch, registry=registry_path, runtime_root=runtime_root,
             goal_id=goal_id, agent_id=agent_id,
         )
-    _project_turn_start_required_reads(
+    turn_start_hook_dispatch = _project_turn_start_required_reads(
         payload,
         turn_start_hook_dispatch,
         available_capabilities=available_capabilities,
@@ -764,6 +770,7 @@ def build_live_quota_should_run_decision(
         turn_instance_id=turn_instance_id,
         registry_path=registry_path,
         runtime_root=runtime_root,
+        context_projector=turn_context_projector,
     )
     hook_dispatch = dispatch_interaction_projection_hooks(interaction_projection_hooks)
     projections = hook_dispatch["projections"]
diff --git a/loopx/control_plane/quota/progressive_context.ts b/loopx/control_plane/quota/progressive_context.ts
index c0479ab174..13c2465704 100644
--- a/loopx/control_plane/quota/progressive_context.ts
+++ b/loopx/control_plane/quota/progressive_context.ts
@@ -488,14 +488,51 @@ function initialContextReadGroups(context: JsonObject, decisionFields: readonly
   include("interaction_contract", "agent_channel", "required_reads");
   const work = ["interaction_contract", "agent_channel", "work_context"];
   include(...work, "instruction");
+  const channel = object(interaction.agent_channel);
+  const sources = object(channel?.work_context)?.sources;
+  const sourceRows = Array.isArray(sources) ? sources.map((source, index) => ({source, index})) : [];
+  const selectedTodo = object(context.selected_todo);
+  const identityAgent = object(context.agent_identity)?.agent_id;
+  const agentId = context.agent_id ?? identityAgent;
+  // Deliver the opted-in decision cue before work, while the candidate or
+  // test can still change. The template and evidence stay on their owner read.
+  const cueStart = pointers.length;
+  if (context.effective_action === EffectiveAction.NORMAL_RUN && context.normal_delivery_allowed === true
+      && typeof selectedTodo?.todo_id === "string" && typeof context.goal_id === "string" && typeof agentId === "string"
+      && (identityAgent === undefined || identityAgent === agentId)) {
+    for (const {source, index} of sourceRows) {
+      const row = object(source), view = object(row?.content), graph = object(view?.graph);
+      if (row?.kind !== "explore_turn_context" || row.capability_id !== "explore"
+          || view?.ok !== true || view.graph_enabled !== true || view.goal_id !== context.goal_id
+          || view.agent_id !== agentId || (view.todo_id !== undefined && view.todo_id !== selectedTodo.todo_id)
+          || typeof object(graph?.work_yardstick_capture)?.guidance !== "string") continue;
+      const graphPath = [...work, "sources", index, "content", "graph"];
+      include(...graphPath, "work_yardstick_capture", "guidance");
+      const measurements = graph?.work_measurements;
+      if (Array.isArray(measurements)) {
+        const current = measurements.findIndex(card => object(card)?.todo_id === selectedTodo.todo_id);
+        if (current >= 0) include(...graphPath, "work_measurements", current);
+      }
+      const cues = graph?.work_yardsticks;
+      if (Array.isArray(cues)) {
+        const current = cues.findIndex(cue => object(cue)?.todo_id === selectedTodo.todo_id);
+        if (current >= 0) include(...graphPath, "work_yardsticks", current);
+      }
+      break;
+    }
+  }
+  // Place the cue after the run decision, before variable-size Goal and work
+  // fields. This also works when the full selection falls back to plain paging.
+  const decisionEnd = pointers.indexOf("/normal_delivery_allowed");
+  if (decisionEnd >= 0 && pointers.length > cueStart) {
+    const cuePointers = pointers.splice(cueStart);
+    pointers.splice(decisionEnd + 1, 0, ...cuePointers);
+  }
   // If the full starting selection needs paging, put the current decision and
   // its read obligations in one independently readable opening. The rest is
   // still required under the original ordering and authority.
   const opening = pointers.splice(0);
   include(...work, "user_todos");
-  const channel = object(interaction.agent_channel);
-  const sources = object(channel?.work_context)?.sources;
-  const sourceRows = Array.isArray(sources) ? sources.map((source, index) => ({source, index})) : [];
   // A complete scoped preference can be larger than a reader page. Visit
   // current narrative sources first, with their availability and provenance,
   // before selecting other sources' metadata: complete-parent coalescing keeps
diff --git a/tests/cli_commands/test_quota_stall_context_reprojection.py b/tests/cli_commands/test_quota_stall_context_reprojection.py
index 7803ff6b78..1eb2e2d9ae 100644
--- a/tests/cli_commands/test_quota_stall_context_reprojection.py
+++ b/tests/cli_commands/test_quota_stall_context_reprojection.py
@@ -89,6 +89,9 @@ def produce():
     args = build_parser().parse_args([
         "--format", "json", "quota", "should-run", "--goal-id", goal,
         "--turn-instance-id", turn,
+        # This case asserts the inline CLI body. Progressive delivery returns
+        # navigation instead and is covered by real captured-reader journeys.
+        "--context-delivery", "off", "--delta-delivery", "off",
     ])
     capture = tmp_path / "capture"
     capture.mkdir()
diff --git a/tests/control_plane/test_work_yardstick_cli.py b/tests/control_plane/test_work_yardstick_cli.py
index b25cc101d5..5560475451 100644
--- a/tests/control_plane/test_work_yardstick_cli.py
+++ b/tests/control_plane/test_work_yardstick_cli.py
@@ -1,4 +1,4 @@
-"""Same-work SQLite journey: evidence → ordinary Turn → real action → readback.
+"""Same-work journey: evidence → ordinary Turn → real action → readback.
 
 The deterministic actor proves transport/adoption attribution, not model quality
 or a causal benchmark gain. No evaluator, credentials or active Goal are used.
@@ -6,13 +6,21 @@
 import hashlib
 from itertools import groupby
 import json
+import shlex
 
+import pytest
 
+from benchmark.runtime.context_delivery import preflight_context_instruction
 from tests.control_plane.test_quota_settlement_cli import (
-    _initialize_fixture_authority, _projected_cli_args, _write_fixture, _run_cli, GOAL_ID, AGENT_ID, TODO_ID,
+    _configure_read_only_todo, _heartbeat_receipt_count, _initialize_fixture_authority, _projected_cli_args,
+    _spend_run_count, _write_fixture, _run_cli, GOAL_ID, AGENT_ID, TODO_ID,
+)
+from loopx.capabilities.explore.result_log import (
+    append_explore_result_event, build_explore_finding_event, build_explore_node_event,
+    explore_result_log_path, load_explore_result_events,
 )
-from loopx.capabilities.explore.result_log import explore_result_log_path, load_explore_result_events
 from loopx.capabilities.explore.turn_context import explore_turn_context
+from loopx.control_plane.turn_driver.decision_capture import read_context_read_reference
 from loopx.presentation.explore_results_api import _result_rows
 
 
@@ -36,8 +44,10 @@ def attachment():
         evidence_refs=["result:reversal"], work_yardstick_decision=decision())
 
 
-def setup(tmp_path, enabled=True, *, active=False):
+def setup(tmp_path, enabled=True, *, active=False, provider="sqlite", write_scope=None, read_only=False):
     project, runtime, path = _write_fixture(tmp_path)
+    if read_only:
+        _configure_read_only_todo(project)
     config = json.loads(path.read_text())
     if active:
         config["goals"][0]["status"] = "active"
@@ -46,8 +56,10 @@ def setup(tmp_path, enabled=True, *, active=False):
     config["goals"][0]["explore_graph"] = {"enabled": True}
     config["goals"][0]["quota"]["allowed_slots"] = 4
     config["goals"][0].setdefault("execution_profile", {})["yardstick_enabled"] = enabled
+    if write_scope is not None:
+        config["goals"][0]["coordination"]["write_scope"] = write_scope
     path.write_text(json.dumps(config))
-    _initialize_fixture_authority(project, runtime, provider="sqlite")
+    _initialize_fixture_authority(project, runtime, provider=provider)
     return project, runtime, path
 
 
@@ -77,6 +89,306 @@ def context(runtime, path, *, detail=True):
     return compact
 
 
+def starting_page(project, runtime, path, turn, saved, monkeypatch):
+    data = saved.read_bytes()
+    assert json.loads(data)["effective_action"] == "normal_run"
+    native = saved.parent / "native"
+    native.mkdir(mode=0o700)
+    env = {"LOOPX_CLI": "loopx", "LOOPX_GOAL_ID": GOAL_ID, "LOOPX_AGENT_ID": AGENT_ID,
+           "LOOPX_CODEX_CONTEXT_DIR": str(native)}
+    prompt = preflight_context_instruction(env, saved, hashlib.sha256(data).hexdigest(), turn[-1])
+    argv = shlex.split(prompt.split("```sh\n", 1)[1].split("\n```", 1)[0])
+    with monkeypatch.context() as patch:
+        patch.setenv("LOOPX_EXECUTION_MODE", "heartbeat")
+        patch.setenv("LOOPX_CONTEXT_DELIVERY", "on")
+        page = run(project, runtime, path, *argv[argv.index("quota"):])
+    assert saved.read_bytes() == data
+    return page
+
+
+def wide_write_scope():
+    return [
+        f"modules/component-{i:03d}/experiments/artifact-encodings/public-validation/"
+        "accepted-input-compatibility/fixture-baseline/ordered-results/recorded-comparison/**"
+        for i in range(100)
+    ]
+
+
+@pytest.mark.parametrize("enabled", [False, True])
+def test_yardstick_cue_reaches_the_starting_read_before_candidate_selection(tmp_path, enabled, monkeypatch):
+    project, runtime, path = setup(tmp_path, enabled, active=True)
+    turn = binding("yardstick-before-choice")
+    capture = tmp_path / "capture"
+    guard = run(project, runtime, path, "quota", "should-run", "--codex-app", *turn,
+                "--decision-output-dir", str(capture), "--context-delivery", "on", "--delta-delivery", "off")
+    assert guard["schema_version"] == "context_capture_navigation_v1"
+    opening = starting_page(project, runtime, path, turn, capture / "decision.json", monkeypatch)["content"]
+    # The independently readable page carries the decision cue. The complete
+    # attachment template stays available on the existing Explore reader.
+    cue_path = "/content/graph/work_yardstick_capture/guidance"
+    assert (cue_path in opening) is enabled
+    assert "/content/graph/work_yardstick_capture/template" not in opening
+    graph = context(runtime, path, detail=False)["graph"]
+    if enabled:
+        assert "use one bounded accepted-outcome comparison" in opening
+        assert graph["work_yardstick_capture"]["optional"] is True
+        assert graph["work_yardstick_capture"]["template"]["proxy"] is None
+    else:
+        assert "work_yardstick_capture" not in graph
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_yardstick_opening_keeps_the_current_cue_with_a_multi_page_write_scope(tmp_path, provider, monkeypatch):
+    scope = wide_write_scope()
+    project, runtime, path = setup(tmp_path, active=True, provider=provider, write_scope=scope)
+    source = tmp_path / "finding.json"
+    source.write_text(json.dumps(attachment()))
+    receipt = refresh(project, runtime, path, "prior-finding", source, "--delivery-boundary", "in_flight_continuation")
+    assert receipt["explore_result_delivery"]["ok"]
+    run(project, runtime, path, *_projected_cli_args(receipt["settlement_owed"]["command"], turn_instance_id="prior-finding"))
+
+    turn = binding("wide-goal-before-choice")
+    capture = tmp_path / "capture"
+    run(project, runtime, path, "quota", "should-run", "--codex-app", *turn,
+        "--decision-output-dir", str(capture), "--context-delivery", "on", "--delta-delivery", "off")
+    saved = capture / "decision.json"
+    original = saved.read_bytes()
+    opening = starting_page(project, runtime, path, turn, saved, monkeypatch)
+    assert opening["offset"] == 0 and not opening["eof"]
+    assert len(json.dumps(opening, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= 12000
+    content = opening["content"]
+    graph = context(runtime, path, detail=False)["graph"]
+    guidance_at = content.index(',"value":', content.index('/graph/work_yardstick_capture/guidance"'))
+    cue_at = content.index(',"value":', content.index('/graph/work_yardsticks/0"'))
+    guidance, _ = json.JSONDecoder().raw_decode(content, guidance_at + len(',"value":'))
+    cue, _ = json.JSONDecoder().raw_decode(content, cue_at + len(',"value":'))
+    assert guidance == graph["work_yardstick_capture"]["guidance"]
+    assert cue == graph["work_yardsticks"][0] and cue["todo_id"] == TODO_ID
+
+    _, reference = read_context_read_reference(opening["reference_file"])
+    assert reference["sha256"] == hashlib.sha256(original).hexdigest()
+    chunks = [content]
+    offset = len(content.encode())
+    page = opening
+    while not page["eof"]:
+        assert page["next_offset"] == offset
+        words = shlex.split(page["next_command"])
+        page = run(project, runtime, path, *words[words.index("quota"):])
+        _, follow = read_context_read_reference(page["reference_file"])
+        assert follow["path"] == reference["path"] and follow["sha256"] == reference["sha256"]
+        assert page["offset"] == offset
+        chunks.append(page["content"])
+        offset += len(page["content"].encode())
+    assert offset == page["total_bytes"]
+    values = {row["pointer"]: row["value"] for row in json.loads("".join(chunks))["selections"]}
+    assert values["/effective_action"] == "normal_run"
+    assert values["/goal_boundary"] == json.loads(original)["goal_boundary"]
+    assert values["/goal_boundary"]["write_scope"] == scope
+    assert saved.read_bytes() == original
+    assert _heartbeat_receipt_count(runtime, turn[-1]) == _spend_run_count(runtime) == 1
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+@pytest.mark.parametrize("enabled", [None, False])
+def test_disabled_yardstick_preserves_exact_first_read_option(tmp_path, provider, enabled, monkeypatch):
+    project, runtime, path = setup(tmp_path, False, active=True, provider=provider)
+    if enabled is None:
+        config = json.loads(path.read_text())
+        del config["goals"][0]["execution_profile"]["yardstick_enabled"]
+        path.write_text(json.dumps(config))
+    turn = binding("off-exact-first-read")
+    capture = tmp_path / "capture"
+    run(project, runtime, path, "quota", "should-run", "--codex-app", *turn,
+        "--decision-output-dir", str(capture), "--context-delivery", "on", "--delta-delivery", "off")
+    saved = capture / "decision.json"
+    original = saved.read_bytes()
+    native = tmp_path / "native"
+    native.mkdir(mode=0o700)
+    env = {"LOOPX_CLI": "loopx", "LOOPX_GOAL_ID": GOAL_ID, "LOOPX_AGENT_ID": AGENT_ID,
+           "LOOPX_CODEX_CONTEXT_DIR": str(native)}
+    prompt = preflight_context_instruction(env, saved, hashlib.sha256(original).hexdigest(), turn[-1])
+    # This existing optional route is an agent decision, not an instruction to
+    # drain an unrelated default opening when the policy is off.
+    assert "For known fields, replace --initial-read with repeated --select pointers" in prompt
+    argv = shlex.split(prompt.split("```sh\n", 1)[1].split("\n```", 1)[0])
+    argv.remove("--initial-read")
+    wanted = ["/effective_action", "/normal_delivery_allowed", "/heartbeat_receipt",
+              "/selected_todo", "/goal_id"]
+    for pointer in wanted:
+        argv.extend(["--select", pointer])
+    with monkeypatch.context() as patch:
+        patch.setenv("LOOPX_EXECUTION_MODE", "heartbeat")
+        patch.setenv("LOOPX_CONTEXT_DELIVERY", "on")
+        page = run(project, runtime, path, *argv[argv.index("quota"):])
+    assert page["eof"]
+    values = {row["pointer"]: row["value"] for row in json.loads(page["content"])["selections"]}
+    assert values == {pointer: json.loads(original)[pointer[1:]] for pointer in wanted}
+    assert saved.read_bytes() == original
+    assert _heartbeat_receipt_count(runtime, turn[-1]) == 1 and _spend_run_count(runtime) == 0
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_first_work_reads_linked_measurements_without_a_yardstick_decision(tmp_path, provider, monkeypatch):
+    project, runtime, path = setup(tmp_path, active=True, provider=provider, write_scope=wide_write_scope())
+    source = tmp_path / "finding.json"
+    finding = attachment()
+    profiles = decision()
+    del finding["work_yardstick_decision"]
+    finding["evidence_refs"] = ["result:a", "result:b"]
+    finding["measurements"] = [
+        {**profiles[profile], "candidate_ref": f"candidate:{candidate}",
+         "evidence_ref": f"result:{candidate}", "value": value}
+        for profile, points in [("authoritative", [("a", 10), ("b", 20)]),
+                                ("proxy", [("a", 9), ("b", 8)])]
+        for candidate, value in points
+    ]
+    source.write_text(json.dumps(finding))
+    receipt = refresh(project, runtime, path, "plain-measurements", source,
+                      "--delivery-boundary", "in_flight_continuation")
+    assert receipt["explore_result_delivery"]["ok"]
+    run(project, runtime, path, *_projected_cli_args(receipt["settlement_owed"]["command"], turn_instance_id="plain-measurements"))
+    events_before = load_explore_result_events(explore_result_log_path(runtime, GOAL_ID))
+    turn = binding("first-linked-comparison")
+    capture = tmp_path / "capture"
+    run(project, runtime, path, "quota", "should-run", "--codex-app", *turn,
+        "--decision-output-dir", str(capture), "--context-delivery", "on", "--delta-delivery", "off")
+    saved = capture / "decision.json"
+    original = saved.read_bytes()
+    page = starting_page(project, runtime, path, turn, saved, monkeypatch)
+    assert page["offset"] == 0 and not page["eof"]
+    assert len(json.dumps(page, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= 12000
+    marker = "/content/graph/work_measurements/0\""
+    start = page["content"].index(',"value":', page["content"].index(marker)) + len(',"value":')
+    comparison, _ = json.JSONDecoder().raw_decode(page["content"], start)
+    assert comparison["todo_id"] == TODO_ID
+    metrics = {row["metric"]: row for row in comparison["measurement_feedback"]}
+    assert metrics["quality"]["best"]["candidate_ref"] == "candidate:b"
+    assert metrics["screen"]["best"]["candidate_ref"] == "candidate:a"
+    assert metrics["quality"]["latest"]["measurement"]["value"] == 20
+    assert comparison["omitted_measurement_feedback"] == 0
+    assert "search_comparison" not in comparison and "authoritative" not in comparison
+    assert "work_yardsticks" not in context(runtime, path, detail=False)["graph"]
+    assert load_explore_result_events(explore_result_log_path(runtime, GOAL_ID)) == events_before
+    assert saved.read_bytes() == original
+    assert _heartbeat_receipt_count(runtime, turn[-1]) == _spend_run_count(runtime) == 1
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+@pytest.mark.parametrize("enabled", [False, True])
+def test_receipt_reentry_keeps_the_selected_work_outside_the_global_context_window(tmp_path, provider, enabled, monkeypatch):
+    project, runtime, path = setup(tmp_path, enabled, active=True, provider=provider)
+    identity = ["--goal-id", GOAL_ID, "--agent-id", AGENT_ID]
+    todo_ids = [TODO_ID]
+    for index in range(4):
+        added = run(project, runtime, path, "todo", "add", "--goal-id", GOAL_ID,
+            "--role", "agent", "--claimed-by", AGENT_ID, "--priority", "P1",
+            "--text", f"Validate independent representation {index}",
+            "--task-class", "advancement_task", "--action-kind", "validate",
+            "--target-key", f"representation-{index}")
+        todo_ids.append(added["todo_id"])
+    log = explore_result_log_path(runtime, GOAL_ID)
+    for index, todo in enumerate(todo_ids):
+        node = f"question-{index}"
+        run(project, runtime, path, "todo", "update", *identity, "--todo-id", todo,
+            "--explore-result-node-ref", node)
+        append_explore_result_event(log, build_explore_node_event(goal_id=GOAL_ID,
+            agent_id=AGENT_ID, node_id=node, title=node, node_kind="question", status="exploring"))
+        evidence = f"result:representation-{index}"
+        append_explore_result_event(log, build_explore_finding_event(goal_id=GOAL_ID,
+            agent_id=AGENT_ID, node_id=node, finding_id=f"finding-{index}",
+            title=f"Observed representation {index}", status="confirmed", evidence_refs=[evidence],
+            measurements=[{**decision()["authoritative"], "value": index,
+                "candidate_ref": f"candidate:{index}", "evidence_ref": evidence}]))
+    global_view = run(project, runtime, path, "explore", "turn-context", *identity)
+    visible = {card["todo_id"] for card in global_view["graph"].get("work_measurements", [])}
+    target = next(todo for todo in todo_ids if todo not in visible)
+    events_before = load_explore_result_events(log)
+    turn_id = "selected-context-reentry"
+    turn = [*identity, "--turn-instance-id", turn_id]
+    command = ["quota", "should-run", "--codex-app", *turn,
+               "--context-delivery", "on", "--delta-delivery", "off"]
+
+    def capture(name, *options):
+        directory = tmp_path / name
+        run(project, runtime, path, *command, *options, "--decision-output-dir", str(directory))
+        saved = directory / "decision.json"
+        return saved, json.loads(saved.read_bytes())
+
+    _, discovery = capture("discovery")
+    assert discovery["interaction_contract"]["cli_channel"]["selection_required"]
+    assert discovery["interaction_contract"]["agent_channel"]["delivery_allowed"] is False
+    _, chosen = capture("chosen", "--todo-id", target)
+    receipts_after_selection = _heartbeat_receipt_count(runtime, turn_id)
+    saved, reentered = capture("reentered")
+    assert chosen["selected_todo"]["todo_id"] == reentered["selected_todo"]["todo_id"] == target
+    assert reentered["effective_action"] == "normal_run" and reentered["normal_delivery_allowed"]
+    assert reentered["interaction_contract"]["agent_channel"]["delivery_allowed"] is True
+    assert chosen["heartbeat_receipt"]["settlement_identity"] == reentered["heartbeat_receipt"]["settlement_identity"]
+    sources = reentered["interaction_contract"]["agent_channel"]["work_context"]["sources"]
+    view = next(source["content"] for source in sources if source.get("kind") == "explore_turn_context")
+    if enabled:
+        assert view["todo_id"] == target
+        assert [card["todo_id"] for card in view["graph"]["work_measurements"]] == [target]
+    else:
+        assert "todo_id" not in view and "work_measurements" not in view["graph"]
+        # The public CLI wraps the same view with invocation-routing metadata.
+        expected = {key: value for key, value in global_view.items() if key != "source_runtime_route"}
+        reference = view["graph"].get("result_writeback_guidance_ref")
+        if reference is not None:
+            # An existing native profile shares the unchanged guidance through
+            # this same capture. Verify the target before comparing its view.
+            assert reference == "$.interaction_contract.cli_channel.settlement_plan.ordered_steps[1].optional_attachments[0].guidance"
+            shared = reentered["interaction_contract"]["cli_channel"]["settlement_plan"]["ordered_steps"][1]["optional_attachments"][0]["guidance"]
+            graph = dict(expected["graph"])
+            assert graph.pop("result_writeback_guidance") == shared
+            expected = {**expected, "graph": {**graph, "result_writeback_guidance_ref": reference}}
+        assert view == expected
+    opening = starting_page(project, runtime, path, turn, saved, monkeypatch)["content"]
+    assert ("/content/graph/work_measurements/0" in opening) is enabled
+    assert target in opening
+    assert load_explore_result_events(log) == events_before
+    assert _heartbeat_receipt_count(runtime, turn_id) == receipts_after_selection
+    assert _spend_run_count(runtime) == 0
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_receipt_settlement_does_not_require_a_closed_work_context(tmp_path, provider):
+    project, runtime, path = setup(tmp_path, active=True, provider=provider, read_only=True)
+    turn_id = "settlement-context-reentry"
+    turn = binding(turn_id)
+    guard = run(project, runtime, path, "quota", "should-run", "--codex-app", *turn)
+    original_identity = guard["heartbeat_receipt"]["settlement_identity"]
+
+    def reenter(expected):
+        counts = _heartbeat_receipt_count(runtime, turn_id), _spend_run_count(runtime)
+        replay = run(project, runtime, path, "quota", "should-run", "--codex-app",
+            "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, "--turn-instance-id", turn_id)
+        channel = replay["interaction_contract"]["agent_channel"]
+        assert replay["normal_delivery_allowed"] is False and channel["delivery_allowed"] is False
+        assert replay["effective_action"] == expected
+        assert replay["heartbeat_receipt"]["settlement_identity"] == original_identity
+        assert channel["work_context"]["complete"]
+        assert all(source.get("error_code") != "context_source_unavailable"
+                   for source in channel["work_context"].get("sources", []))
+        assert (_heartbeat_receipt_count(runtime, turn_id), _spend_run_count(runtime)) == counts
+
+    run(project, runtime, path, "todo", "complete", *turn, "--claimed-by", AGENT_ID,
+        "--evidence", "artifact:verified-synthetic-evidence")
+    reenter("unsettled_host_turn_recovery")
+    archived = run(project, runtime, path, "todo", "archive-completed", "--goal-id", GOAL_ID,
+        "--max-active-done", "0", "--execute")
+    assert archived["moved_count"] == 1
+    reenter("unsettled_host_turn_recovery")
+    run(project, runtime, path, "refresh-state", *turn,
+        "--classification", "validated_change", "--delivery-batch-scale", "single_surface",
+        "--delivery-outcome", "outcome_progress", "--no-global-sync", "--suppress-external-sinks")
+    reenter("unsettled_host_turn_recovery")
+    run(project, runtime, path, "quota", "spend-slot", *turn,
+        "--slots", "1", "--source", "heartbeat", "--execute")
+    reenter("heartbeat_settled_skip")
+    assert _heartbeat_receipt_count(runtime, turn_id) == _spend_run_count(runtime) == 1
+
+
 def test_measured_yardstick_restarts_comparison_when_the_acceptance_basis_changes(tmp_path):
     project, runtime, path = setup(tmp_path, active=True)
     source = tmp_path / "finding.json"
@@ -114,7 +426,7 @@ def record(turn, acceptance, candidates, previous=None):
     assert comparison["preferred_candidate_ref"] == "candidate:new-d"
 
 
-def test_measured_yardstick_changes_next_work_from_local_screening_to_a_whole_goal_comparison(tmp_path):
+def test_measured_yardstick_changes_next_work_from_local_screening_to_a_whole_goal_comparison(tmp_path, monkeypatch):
     project, runtime, path = setup(tmp_path, active=True)
     acceptance = tmp_path / "acceptance.json"
     acceptance.write_text(json.dumps(dict(outcome_id="encoded-artifact", owner_agent_id=AGENT_ID,
@@ -160,7 +472,10 @@ def dictionary_encode(items):
     assert first["explore_result_delivery"]["ok"]
     run(project, runtime, path, *_projected_cli_args(first["settlement_owed"]["command"], turn_instance_id="turn-pair"))
 
-    guard = run(project, runtime, path, "quota", "should-run", "--codex-app", *binding("turn-compare"))
+    compare_turn = binding("turn-compare")
+    capture = tmp_path / "compare"
+    guard = run(project, runtime, path, "quota", "should-run", "--codex-app", *compare_turn,
+                "--decision-output-dir", str(capture))
     compact = run(project, runtime, path, "explore", "turn-context", "--goal-id", GOAL_ID, "--agent-id", AGENT_ID,
                   "--todo-id", TODO_ID)["graph"]["work_yardsticks"][0]
     assert compact["comparison_candidate_ref"] == "candidate:dictionary"
@@ -168,6 +483,13 @@ def dictionary_encode(items):
     assert "different mechanism or a compatible combination" in compact["comparison_cue"]
     assert "full implementation and validation cost" in compact["comparison_cue"]
     assert compact["comparison_cue"] in json.dumps(guard)
+    opening = starting_page(project, runtime, path, compare_turn, capture / "decision.json", monkeypatch)["content"]
+    # A page may end in a later entry. Decode the complete cue before that
+    # boundary without relying on any subsequent context read.
+    cue_start = opening.index(',"value":', opening.index("/content/graph/work_yardsticks/"))
+    delivered, _ = json.JSONDecoder().raw_decode(opening, cue_start + len(',"value":'))
+    assert delivered["todo_id"] == TODO_ID
+    assert delivered["comparison_candidate_ref"] == "candidate:dictionary"
     detailed = run(project, runtime, path, "explore", "turn-context", "--goal-id", GOAL_ID, "--agent-id", AGENT_ID,
                    "--todo-id", TODO_ID, "--result-node", compact["node_id"])["graph"]["work_yardsticks"][0]
     assert detailed["next_action"] == choice["next_action"]
@@ -178,7 +500,7 @@ def dictionary_encode(items):
 
     # Deterministic consumer. Combine the evidenced baseline with an independent
     # representation change and verify the original end-to-end outcome.
-    selected_base = {"candidate:dictionary": base}[compact["comparison_candidate_ref"]]
+    selected_base = {"candidate:dictionary": base}[delivered["comparison_candidate_ref"]]
     runs = [[index, len(list(group))] for index, group in groupby(selected_base["indices"])]
     artifact = project / "combined-encoding.json"
     artifact.write_bytes(encode({"keys": selected_base["keys"], "runs": runs}))
diff --git a/tests/control_plane_ts/context_selection.test.ts b/tests/control_plane_ts/context_selection.test.ts
index 43ac3f534b..6a189326aa 100644
--- a/tests/control_plane_ts/context_selection.test.ts
+++ b/tests/control_plane_ts/context_selection.test.ts
@@ -146,6 +146,45 @@ function selectedFacts(context: unknown, selections: {path: (string | number)[]}
   return [...facts].sort();
 }
 
+test("only an admitted same-source Yardstick can precede the complete Goal boundary", () => {
+  const source = {kind: "explore_turn_context", capability_id: "explore", content: {
+    ok: true, graph_enabled: true, goal_id: "g", agent_id: "a", todo_id: "current",
+    graph: {
+      work_yardstick_capture: {guidance: "compare the accepted outcome", template: "optional detail"},
+      work_yardsticks: [{todo_id: "other", cue: "unrelated experiment"}, {todo_id: "current", cue: "current experiment"}],
+      work_measurements: [{todo_id: "other", measurement_feedback: ["foreign measurements"]},
+        {todo_id: "current", measurement_feedback: ["linked comparisons without a decision"]}],
+    },
+  }};
+  const context = {
+    goal_id: "g", agent_identity: {agent_id: "a"}, effective_action: "normal_run",
+    normal_delivery_allowed: true, selected_todo: {todo_id: "current"},
+    goal_boundary: {rule: "Use only the authorized scope.", write_scope: ["work/**"]},
+    interaction_contract: {schema_version: "loopx_interaction_contract_v0",
+      agent_channel: {work_context: {sources: [source]}}},
+  };
+  const initial = (capture = context) => (evaluateTurnEnvelope({
+    operation: "context_select", context: capture, initial_read: true,
+  }).selections as {pointer: string; path: (string | number)[]}[]);
+  const rows = initial();
+  const choice = rows.findIndex(row => row.pointer.endsWith("/work_yardsticks/1"));
+  assert.ok(choice > rows.findIndex(row => row.pointer === "/normal_delivery_allowed"));
+  assert.ok(choice < rows.findIndex(row => row.pointer === "/goal_boundary"));
+  const measurements = rows.findIndex(row => row.pointer.endsWith("/work_measurements/1"));
+  assert.ok(measurements > rows.findIndex(row => row.pointer === "/normal_delivery_allowed"));
+  assert.ok(measurements < rows.findIndex(row => row.pointer === "/goal_boundary"));
+  assert.ok(!selectedFacts(context, rows).some(fact => /unrelated experiment|foreign measurements|optional detail/.test(fact)));
+  for (const binding of [{goal_id: "another-goal"}, {agent_id: "another-agent"}, {todo_id: "other"}]) {
+    const capture = structuredClone(context);
+    Object.assign(capture.interaction_contract.agent_channel.work_context.sources[0].content, binding);
+    assert.ok(initial(capture).every(row => !/work_yardstick|work_measurements/.test(row.pointer)));
+  }
+  for (const gate of [{normal_delivery_allowed: false}, {effective_action: "autonomous_replan_required"},
+    {agent_id: "a", agent_identity: {agent_id: "another-agent"}}]) {
+    assert.ok(initial({...context, ...gate}).every(row => !/work_yardstick|work_measurements/.test(row.pointer)));
+  }
+});
+
 test("initial navigation coalesces complete branches without changing selected facts or explicit readers", () => {
   const context = {facts: [{"a/b": "Keep the final stop.", "~": null}, {allowed: false, evidence: []}],
     single: {only: "A single address need not grow a container."}, unread: {history: "optional body"}};
diff --git a/tests/control_plane_ts/explore_turn_context.test.ts b/tests/control_plane_ts/explore_turn_context.test.ts
index d2205cc8b3..f582a7171e 100644
--- a/tests/control_plane_ts/explore_turn_context.test.ts
+++ b/tests/control_plane_ts/explore_turn_context.test.ts
@@ -5,6 +5,81 @@ import type {JsonObject} from "../../loopx/control_plane/effect_program.ts";
 import {projectStrideExploreBasis, strideReviewMismatch} from "../../loopx/control_plane/work_items/stride_policy.ts";
 import {projectExploreTurnContext} from "../../loopx/control_plane/capabilities/explore_turn_context.ts";
 
+test("initial work measurement feedback preserves linked scopes without choosing an outcome profile", () => {
+  const profile = {metric: "quality", unit: "points", higher_is_better: true,
+    comparison_scope: "inputs:shared", input_revision: "v1", evaluation_ref: "judge:public"};
+  const finding = (id: string, node: string, order: number, candidate: string, value: number | null,
+    changes: JsonObject = {}) => ({finding_id: id, node_id: node, recorded_order: order,
+    measurements: [{...profile, ...changes, candidate_ref: candidate, evidence_ref: `result:${id}`, value}]});
+  const todo = {todo_id: "todo_current", role: "agent", status: "open", done: false, claimed_by: "worker",
+    explore_result_node_refs: ["shared", "local"]};
+  const request = {goal_id: "research", agent_id: "worker", route: ["loopx"], yardstick_enabled: true,
+    graph_enabled: true, harness_gate: {enabled: false}, work_todos: [todo], plan: {},
+    projection: {nodes: [], findings: [
+      finding("first", "shared", 1, "candidate:a", 10),
+      finding("second", "shared", 2, "candidate:b", 20),
+      finding("foreign", "unlinked", 3, "candidate:foreign", 1000),
+      finding("local-a", "local", 4, "candidate:a", 9, {metric: "screen"}),
+      finding("local-b", "local", 5, "candidate:b", 8, {metric: "screen"}),
+      finding("unknown", "local", 6, "candidate:c", null, {input_revision: "v2"}),
+      finding("other-unit", "local", 7, "candidate:c", 30, {unit: "percent"}),
+    ]}};
+  const original = structuredClone(request);
+  const graph = projectExploreTurnContext(request).graph as JsonObject;
+  const card = graph.work_measurements[0];
+  assert.equal(card.todo_id, todo.todo_id);
+  assert.equal(card.measurement_feedback.length, 3);
+  assert.equal(card.omitted_measurement_feedback, 1);
+  const outcome = card.measurement_feedback.find(row => row.node_id === "shared");
+  assert.equal(outcome.best.candidate_ref, "candidate:b");
+  assert.equal(outcome.best.value, 20);
+  assert.equal(outcome.latest.finding_id, "second");
+  assert.equal(outcome.input_revision, "v1");
+  const unknown = card.measurement_feedback.find(row => row.input_revision === "v2");
+  assert.equal(unknown.latest.measurement.value, null);
+  assert.equal(unknown.best, null);
+  assert.equal(unknown.improvement, null);
+  assert.ok(card.measurement_feedback.every(row => row.node_id !== "unlinked"));
+  assert.ok(!Object.hasOwn(card, "authoritative") && !Object.hasOwn(card, "search_comparison"));
+  assert.ok(!Object.hasOwn(graph, "work_yardsticks"));
+  assert.ok(card.details_command.includes("--todo-id") && card.details_command.includes(todo.todo_id));
+  assert.deepEqual(request, original);
+  for (const enabled of [false, undefined]) {
+    const off = projectExploreTurnContext({...request, yardstick_enabled: enabled}).graph as JsonObject;
+    assert.ok(!Object.hasOwn(off, "work_measurements"));
+    assert.deepEqual(off.measurement_feedback, graph.measurement_feedback);
+  }
+  for (const changes of [{claimed_by: "another"}, {done: true}, {status: "blocked"}, {explore_result_node_refs: []}]) {
+    const unavailable = projectExploreTurnContext({...request, work_todos: [{...todo, ...changes}]}).graph as JsonObject;
+    assert.ok(!Object.hasOwn(unavailable, "work_measurements"));
+  }
+  const scoped = projectExploreTurnContext({...request, result_node: "unlinked"}).graph as JsonObject;
+  assert.ok(!Object.hasOwn(scoped, "work_measurements"));
+  const shared = projectExploreTurnContext({...request,
+    work_todos: [todo, {...todo, todo_id: "todo_shared"}]}).graph as JsonObject;
+  assert.deepEqual(shared.work_measurements.map(row => row.todo_id), ["todo_current", "todo_shared"]);
+  assert.deepEqual(shared.work_measurements[0].measurement_feedback, shared.work_measurements[1].measurement_feedback);
+  const tied = structuredClone(request);
+  tied.projection.findings[1].measurements[0].value = 10;
+  const tiedGraph = projectExploreTurnContext(tied).graph as JsonObject;
+  const tie = tiedGraph.work_measurements[0].measurement_feedback.find(row => row.node_id === "shared");
+  assert.equal(tie.best.candidate_ref, "candidate:a");
+  assert.equal(tie.latest.measurement.candidate_ref, "candidate:b");
+  assert.equal(tie.improvement, 0);
+  const declared: JsonObject = {...request, result_node: "local", projection: {...request.projection,
+    findings: [...request.projection.findings, {finding_id: "declared", node_id: "shared", recorded_order: 8,
+      evidence_refs: [], work_yardstick: {binding: {goal_id: "research", agent_id: "worker",
+        todo_id: todo.todo_id, turn_instance_id: "turn:declared"}, decision: {
+        schema_version: "work_yardstick_decision_v0", target: {summary: "Improve the accepted outcome",
+          scope: "Original work", acceptance_ref: "acceptance:original"}, authoritative: profile,
+        proxy: null, counterevidence_refs: [], prior_finding_id: null,
+        next_action: {method_id: "method:retained", reason: "Continue the recorded choice", cost_bound: "Existing budget"},
+      }}}]}};
+  const differentQuestion = projectExploreTurnContext(declared).graph as JsonObject;
+  assert.ok(!Object.hasOwn(differentQuestion, "work_measurements"));
+  assert.ok(differentQuestion.measurement_feedback.every(row => row.node_id === "local"));
+});
+
 test("recent questions use UTC instants, preserve microseconds and retain legacy timestamps", () => {
   for (const timestamps of [
     ["2026-01-02T08:00:00+08:00", "2026-01-02T00:01:00Z", "2026-01-01T19:02:00-05:00"],