refactor(todos): compose canonical quota sources and bound responses

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

loopx/control_plane/effect_runtime.py

@@ -169,6 +169,12 @@ def remember_todo_context(self, key: bytes, result: bytes) -> None:
 )
 
 
+def todo_projection_request_identity() -> object | None:
+    """Bind a source codec to its current logical request, never to a TTL."""
+    state = _REQUEST_RUNTIME_REVISION.get()
+    return state if state is not None and not state._closed else None
+
+
 class EffectRuntimeRemoteError(RuntimeError):
     """A typed exception returned by the managed TypeScript runtime."""
 
@@ -1601,6 +1607,17 @@ def effect_runtime_result(
     large_local_snapshot: bool = False,
 ) -> Any:
     state = _REQUEST_RUNTIME_REVISION.get() if method == "todo.context.page" else None
+    def validate_result(result: Any) -> Any:
+        # A caller must explicitly request a bundle. A request scope or a small
+        # source does not establish that it will consume additional views.
+        if set(params) == {"summary_display_views"} and (not isinstance(result, dict) or any(
+            not isinstance(result.get(view), list)
+            or any(not isinstance(row, dict) for row in result[view])
+            for view in ("summary", "planning", "terminal")
+        )):
+            raise RuntimeError("invalid TypeScript Todo source views")
+        return result
+
     key = None
     if state is not None:
         # Preserve JSON order and scalar types, including bool/int, signed zero
@@ -1613,14 +1630,15 @@ def effect_runtime_result(
         if key is not None:
             prior = state.todo_context_result(key)
             if prior is not None:
-                return json.loads(prior)  # A caller cannot mutate another view.
+                return validate_result(json.loads(prior))  # A caller cannot mutate another view.
     result = effect_runtime_request(
         method,
         params,
         timeout=timeout,
         retry_safe=retry_safe,
         large_local_snapshot=large_local_snapshot,
     ).get("result")
+    validate_result(result)
     if state is not None and key is not None and isinstance(result, dict):
         try:
             encoded = json.dumps(result, ensure_ascii=False, allow_nan=False,

loopx/control_plane/effect_runtime_server.ts

@@ -15,7 +15,7 @@ import {
 import { atomicWriteJson, withFileMutationLock } from "./effect_runtime_io.ts";
 import { retireRuntime } from "./effect_runtime_compile_cache.ts";
 import { sqliteRuntimeIdentity } from "./coordination/sqlite_runtime.ts";
-import {openPrivateResponseSink, readPrivateJsonSnapshot, writePrivateResponse} from "./effect_runtime_snapshot.ts";
+import {MAX_INLINE_RESPONSE_BYTES, openPrivateResponseSink, readPrivateJsonSnapshot, writePrivateResponse} from "./effect_runtime_snapshot.ts";
 import {
   requireJsonObject as requiredObject,
   requireNonEmptyString as requiredString,
@@ -26,7 +26,6 @@ const RESPONSE_SCHEMA = "loopx_effect_runtime_response_v1";
 const INFO_SCHEMA = "loopx_effect_runtime_info_v0";
 const STARTUP_ERROR_SCHEMA = "loopx_effect_runtime_startup_error_v0";
 const MAX_REQUEST_BYTES = 8 * 1024 * 1024;
-const MAX_INLINE_RESPONSE_BYTES = 8 * 1024 * 1024;
 // Explicit opt-in: ordinary effects retain a bounded request/response wire.
 const LOCAL_SNAPSHOT_METHODS = new Set(localSnapshotMethods);
 const DEFAULT_IDLE_MS = 5 * 60 * 1_000;

loopx/control_plane/effect_runtime_snapshot.ts

@@ -11,6 +11,7 @@ import {requireJsonObject, requireNonEmptyString} from "./runtime_decode.ts";
 import { BARE_SHA256_PATTERN } from "./content_digest.ts";
 
 export const MAX_LOCAL_SNAPSHOT_BYTES = 64 * 1024 * 1024;
+export const MAX_INLINE_RESPONSE_BYTES = 8 * 1024 * 1024;
 
 async function privateFile(path: string, flags: number): Promise<FileHandle> {
   if (!isAbsolute(path) || !basename(dirname(path)).startsWith("loopx-effect-")) {

loopx/control_plane/goals/goal_vision_wait.py

@@ -79,6 +79,20 @@ def build_goal_vision_wait_state(
     """
 
     summary = agent_todo_summary if isinstance(agent_todo_summary, dict) else {}
+    from ..todos.summary_item import quota_vision_wait_basis
+    basis = quota_vision_wait_basis(summary, source_items, agent_id) if lineage_source_items is None else None
+    if basis is not None:
+        result = effect_runtime_result("goal.vision_wait.project", {
+            "agent_todo_summary": {"vision_wait_states": summary.get("vision_wait_states"),
+                "current_agent_blocker_items": basis["current_agent_blocker_items"]},
+            "agent_id": agent_id, "acceptance_gaps": acceptance_gaps or [],
+            "selectable_advancement_count": selectable_advancement_count, "wait_basis": basis,
+        }, large_local_snapshot=True)
+        if result is not None and (not isinstance(result, dict)
+                or result.get("schema_version") != GOAL_VISION_WAIT_STATE_SCHEMA_VERSION
+                or result.get("state") != "waiting"):
+            raise RuntimeError("TypeScript vision wait state shape mismatch")
+        return result
     planning_input = summary
     if source_items is not None:
         planning_input = {

loopx/control_plane/goals/vision_wait_coverage.ts

@@ -159,10 +159,13 @@ function projectVisionWaitFromFacts(input: {request: JsonObject; gaps: JsonObjec
   const summary = request.agent_todo_summary === null ? {} : requireJsonObject(request.agent_todo_summary, "agent_todo_summary");
   const cachedWait = cachedWaitProof(summary, agentId, causalIds);
   if (cachedWait) return cachedWait;
-  const resumePlan = projectTodoResumePlanning(requireJsonObject(request.resume_planning_request, "resume_planning_request"));
+  const basis = request.wait_basis == null ? null : requireJsonObject(request.wait_basis, "wait basis");
+  const resumePlan = basis === null ? projectTodoResumePlanning(requireJsonObject(request.resume_planning_request, "resume_planning_request")) :
+    requireJsonObject(basis.resume_planning, "prepared resume planning");
   const blockers = currentAgentBlockers(summary.current_agent_blocker_items, agentId);
   const candidates = Array.isArray(resumePlan.blocked_successor_items) ? resumePlan.blocked_successor_items as JsonObject[] : [];
-  const lineageRows = Array.isArray(request.lineage_source_items) ? request.lineage_source_items : candidates;
+  const lineageRows = Array.isArray(request.lineage_source_items) ? request.lineage_source_items :
+    basis !== null && Array.isArray(basis.lineage_source_items) ? basis.lineage_source_items : candidates;
   const coverage = projectWaitCoverage(causalIds, candidates, blockers, lineageRows);
   if (coverage.covered !== true) return null;
   const waitSelection = selectedWaitItems(blockers, candidates, new Set(coverage.witness_todo_ids as string[]));

loopx/control_plane/quota/should_run_prepare.py

@@ -98,9 +98,7 @@
     todo_item_task_class as projection_todo_item_task_class,
 )
 from ..todos.quota_summary import (
-    select_planning_inventory_source_items,
-    select_quota_todo_source_items,
-    select_quota_todo_summary,
+    select_quota_todo_contexts,
     select_task_orchestration_authority_items,
 )
 from ..todos.summary_item import compact_todo_summary_item, compact_todo_summary_items
@@ -237,6 +235,7 @@ def _prepare_quota_replan_context(
     project_asset: dict[str, Any],
     user_todo_summary: dict[str, Any] | None,
     agent_todo_summary: dict[str, Any] | None,
+    agent_todo_terminal_source_items: list[dict[str, Any]],
     authoritative_fallback_todo_items: FallbackTodoSource,
     effective_available_capabilities: Any,
     work_lane_contract: dict[str, Any] | None,
@@ -251,11 +250,6 @@ def _prepare_quota_replan_context(
     replan_explore_reader: Callable[[str, str | None], Mapping[str, Any] | None] | None,
 ) -> tuple[dict[str, Any], dict[str, Any] | None]:
     """Project the current frontier and contextualize its receipt-bound review."""
-    replan_todo_items = select_planning_inventory_source_items(
-        item.get("agent_todos"),
-        project_asset.get("agent_todos") if project_asset else None,
-        include_terminal=True,
-    )
     goal_frontier_context = build_goal_frontier_projection_context_from_status(
         replan_explore_reader=replan_explore_reader,
         goal_id=safe_goal_id,
@@ -265,7 +259,7 @@ def _prepare_quota_replan_context(
         project_asset=project_asset,
         user_todo_summary=user_todo_summary,
         agent_todo_summary=agent_todo_summary,
-        agent_todo_source_items=replan_todo_items,
+        agent_todo_source_items=agent_todo_terminal_source_items,
         fallback_todo_source_items=authoritative_fallback_todo_items,
         available_capabilities=effective_available_capabilities,
         work_lane_contract=work_lane_contract,
@@ -296,7 +290,7 @@ def _prepare_quota_replan_context(
             source_status=status_payload, stride_strategy=goal_frontier_context["stride_strategy"],
             explore_reader=replan_explore_reader,
             goal_acceptance_contract=(agent_todo_summary or {}).get("goal_acceptance_contract"),
-            todo_items=replan_todo_items,
+            todo_items=agent_todo_terminal_source_items,
         )
     return goal_frontier_context, replan_obligation
 
@@ -701,40 +695,29 @@ def _prepare_quota_should_run_item(
         item=item, project_asset=project_asset, available=available_capabilities,
     )
     effective_available_capabilities = availability["effective"]
-    user_todo_summary = select_quota_todo_summary(
+    user_todo_summary, agent_todo_summary, agent_todo_display_views, user_todo_source_items = select_quota_todo_contexts(
         item.get("user_todos"),
         project_asset.get("user_todos") if project_asset else None,
-        agent_identity=agent_identity,
-        filter_user_gate_blocks_agent=True,
-        available_capabilities=effective_available_capabilities,
-    )
-    agent_todo_summary = select_quota_todo_summary(
         item.get("agent_todos"),
         project_asset.get("agent_todos") if project_asset else None,
         agent_identity=agent_identity,
         available_capabilities=effective_available_capabilities,
     )
-    user_todo_source_items = select_quota_todo_source_items(
-        item.get("user_todos"),
-        project_asset.get("user_todos") if project_asset else None,
-    )
-    agent_todo_source_items = select_quota_todo_source_items(
-        item.get("agent_todos"),
-        project_asset.get("agent_todos") if project_asset else None,
-    )
-    agent_todo_planning_source_items = select_planning_inventory_source_items(
-        item.get("agent_todos"),
-        project_asset.get("agent_todos") if project_asset else None,
-    )
+    # Only this Agent source is consumed as summary, planning and retained
+    # terminal rows below. User-source and general display calls stay single-view.
+    agent_todo_source_items = agent_todo_display_views["summary"]
+    agent_todo_planning_source_items = agent_todo_display_views["planning"]
     task_orchestration_agent_items = select_task_orchestration_authority_items(
         item.get("agent_todos"),
         project_asset.get("agent_todos") if project_asset else None,
         role="agent",
+        fallback_source_items=agent_todo_source_items,
     )
     task_orchestration_user_blockers = select_task_orchestration_authority_items(
         item.get("user_todos"),
         project_asset.get("user_todos") if project_asset else None,
         role="user",
+        fallback_source_items=user_todo_source_items,
     )
     agent_scoped_user_todo_override = _agent_scoped_user_todo_override(
         state=state,
@@ -986,6 +969,7 @@ def _prepare_quota_should_run_item(
         project_asset=project_asset,
         user_todo_summary=user_todo_summary,
         agent_todo_summary=agent_todo_summary,
+        agent_todo_terminal_source_items=agent_todo_display_views["terminal"],
         authoritative_fallback_todo_items=authoritative_fallback_todo_items,
         effective_available_capabilities=effective_available_capabilities,
         work_lane_contract=work_lane_contract,

loopx/control_plane/todos/context_projection.ts

@@ -24,6 +24,19 @@ const CONTEXT_FIELDS = [
 
 /** Select before redaction; never convert nested scopes or booleans to prose. */
 export function projectTodoContextPage(input: JsonObject): JsonObject {
+  if (input.summary_display_views !== undefined) {
+    if (Object.keys(input).length !== 1) throw new EffectRuntimeRequestError("summary display views cannot be combined with other views");
+    const source = requireJsonObject(input.summary_display_views, "summary display views");
+    if (Object.keys(source).some(key => key !== "source" && key !== "handoff_gates")) {
+      throw new EffectRuntimeRequestError("invalid summary display views fields");
+    }
+    requireJsonObject(source.source, "summary source");
+    return {
+      summary: projectTodoSummaryDisplay(source),
+      planning: projectTodoSummaryDisplay({source: source.source, include_terminal: false}),
+      terminal: projectTodoSummaryDisplay({source: source.source, include_terminal: true}),
+    };
+  }
   if (input.summary_display !== undefined) {
     if (Object.keys(input).length !== 1) throw new EffectRuntimeRequestError("summary display cannot be combined with other views");
     return {items: projectTodoSummaryDisplay(requireJsonObject(input.summary_display, "summary display"))};
@@ -188,14 +201,14 @@ const SUMMARY_FIELDS = [
   "route_key", "completed_at", "updated_at", "completion_turn_key", "superseded_by", "handoff_note",
 ];
 
-const SUMMARY_SOURCES = [
+export const SUMMARY_SOURCES = [
   "active_next_action_items", "active_next_action_executable_items", "first_open_items", "backlog_items", "unclaimed_priority_open_items",
   "claimed_open_items", "claimed_advancement_open_items", "claimed_monitor_open_items", "monitor_open_items", "current_agent_claimed_open_items",
   "current_agent_claimed_advancement_items", "current_agent_claimed_monitor_items", "blocker_items", "resume_blocked_items", "monitor_blocked_resume_candidates",
   "current_agent_monitor_blocked_resume_candidates", "unclaimed_monitor_blocked_resume_candidates", "items",
 ];
 
-const PLANNING_SOURCES = ["items", "deferred_items", "blocker_items", "monitor_open_items"];
+export const PLANNING_SOURCES = ["items", "deferred_items", "blocker_items", "monitor_open_items"];
 
 function compactSummaryRecord(item: JsonObject, body: unknown): JsonObject {
   const row: JsonObject = {index: item.index ?? null, text: body};