Reuse canonical Todo reads across quota and refresh projections

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

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，把前次
 提议方法连到实际方法、候选和证据，也允许有据采用不同方法。

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

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 {

loopx/chat_server.py

@@ -1727,7 +1727,8 @@ def do_GET(self) -> None:
             SSH_HOST_CATALOG_PATH: self._ssh_hosts,
         }
         if path in get_dispatch:
-            return get_dispatch[path]()
+            with effect_runtime_request_scope():
+                return get_dispatch[path]()
         if self._dispatch_goal_result_read(path):
             return
         setup_parts = path.strip("/").split("/")

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,

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"] = {

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,
             )

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,

loopx/control_plane/agents/profile.py

@@ -2,6 +2,7 @@
 
 import re
 from fnmatch import fnmatchcase
+from functools import lru_cache
 from typing import Any, Mapping
 
 from ..runtime.public_safety import public_safe_compact_text
@@ -200,18 +201,35 @@ def agent_profile_candidate_rank(
         return 1
     action_kind = normalize_todo_action_kind(item.get("action_kind")) or ""
     avoided = agent_profile.get("avoid_action_kinds")
-    if action_kind and isinstance(avoided, list) and any(
+    avoid = bool(action_kind and isinstance(avoided, list) and any(
         fnmatchcase(action_kind, str(pattern)) for pattern in avoided
-    ):
-        return 2
+    ))
     preferred = agent_profile.get("preferred_action_kinds")
-    if action_kind and isinstance(preferred, list) and any(
+    prefer = bool(action_kind and isinstance(preferred, list) and any(
         fnmatchcase(action_kind, str(pattern)) for pattern in preferred
-    ):
-        return 0
+    ))
     default_task_classes = agent_profile.get("default_task_classes")
-    if isinstance(default_task_classes, list):
-        task_class = str(item.get("task_class") or "").strip().lower()
-        if task_class and task_class in default_task_classes:
-            return 0
-    return 1
+    task_class = str(item.get("task_class") or "").strip().lower()
+    default_class = bool(task_class and isinstance(default_task_classes, list)
+                         and task_class in default_task_classes)
+    from ..effect_runtime import _runtime_fingerprint_for_request
+    return _typed_profile_rank(avoid, prefer, default_class, _runtime_fingerprint_for_request())
+
+
+@lru_cache(maxsize=16)
+def _typed_profile_rank(avoid: bool, prefer: bool, default_class: bool, revision: str) -> int:
+    """Legacy fnmatch IO stays here; advisory rank precedence has one TS owner.
+
+    Memoize the eight immutable Boolean combinations at their runtime revision,
+    never item, profile, source, authority or an execution decision.
+    """
+    from ..effect_runtime import effect_runtime_result
+
+    result = effect_runtime_result("todo.quota_planning.project", {
+        "schema_version": "todo_profile_rank_request_v0",
+        "facts": {"avoid": avoid, "prefer": prefer, "default_class": default_class},
+    })
+    rank = result.get("rank") if isinstance(result, dict) else None
+    if type(rank) is not int or rank not in (0, 1, 2):
+        raise RuntimeError("invalid typed advisory profile rank")
+    return rank