perf(status): eliminate redundant registry reads in status hotpath

loopx/contract.py

@@ -889,6 +889,7 @@ def check_contract(
     activation_state_filter: GoalActivationState | str | None = None,
     include_public_boundary_scan: bool = True,
     history_audit: RunHistoryAudit | None = None,
+    registry: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     error_diagnostics: list[dict[str, Any]] = []
     warnings: list[str] = []
@@ -936,7 +937,8 @@ def add_global_error(code: str, message: str) -> None:
         for risk in boundary_payload.get("risks") or []:
             add_global_error("registry_boundary_risk", f"registry boundary risk: {risk}")
 
-    registry = load_registry(registry_path)
+    if registry is None:
+        registry = load_registry(registry_path)
     todo_contract_diagnostics, checked_user_gates = (
         _active_state_todo_contract_diagnostics(
             registry,

loopx/control_plane/runtime/runtime_projection_route.py

@@ -501,8 +501,10 @@ def _source_routes_for_registry(
     goal_id: str | None,
     activation_state_filter: GoalActivationState | str | None = None,
     source_registry_read_timeout_seconds: float = SOURCE_REGISTRY_READ_TIMEOUT_SECONDS,
+    registry: dict[str, Any] | None = None,
 ) -> list[tuple[Path, Path, str, str | None]]:
-    registry = load_registry(registry_path)
+    if registry is None:
+        registry = load_registry(registry_path)
     is_global = bool(registry.get("registry_role") == "global-local") or _same_path(
         registry_path,
         global_registry_path(runtime_root),
@@ -598,6 +600,7 @@ def collect_runtime_projection_route_diagnostics(
     goal_id: str | None = None,
     activation_state_filter: GoalActivationState | str | None = None,
     source_registry_read_timeout_seconds: float = SOURCE_REGISTRY_READ_TIMEOUT_SECONDS,
+    registry: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     items: list[dict[str, Any]] = []
     source_routes = _source_routes_for_registry(
@@ -606,8 +609,10 @@ def collect_runtime_projection_route_diagnostics(
         goal_id=goal_id,
         activation_state_filter=activation_state_filter,
         source_registry_read_timeout_seconds=source_registry_read_timeout_seconds,
+        registry=registry,
     )
-    registry = load_registry(registry_path)
+    if registry is None:
+        registry = load_registry(registry_path)
     registry_is_global = bool(registry.get("registry_role") == "global-local") or _same_path(
         registry_path,
         global_registry_path(runtime_root),

loopx/control_plane/status/collection.py

@@ -116,6 +116,7 @@ def collect_status(
         status_include_runtime_goals=include_runtime_goals,
         activation_state_filter=activation_filter,
         agent_lane_id=agent_lane_id,
+        registry=registry,
     )
     history = history_collection.status_history
     contract = context.check_contract(
@@ -127,6 +128,7 @@ def collect_status(
         include_public_boundary_scan=include_public_boundary_scan,
         activation_state_filter=activation_filter,
         history_audit=history_collection.contract_audit,
+        registry=registry,
     )
     contract = project_contract_health_for_goal(contract, goal_id=goal_filter)
     queue = context.build_attention_queue(
@@ -157,12 +159,14 @@ def collect_status(
     promotion_gate = context.build_promotion_gate(
         registry_path=registry_path,
         runtime_root_override=str(runtime_root),
+        registry=registry,
     )
     runtime_projection_routes = collect_runtime_projection_route_diagnostics(
         registry_path=registry_path,
         runtime_root=runtime_root,
         goal_id=goal_filter,
         activation_state_filter=activation_filter,
+        registry=registry,
     )
     runtime_projection_route_health = {
         "healthy": (

loopx/history.py

@@ -307,14 +307,16 @@ def collect_history(
     include_runtime_goals: bool = True,
     activation_state_filter: GoalActivationState | str | None = None,
     agent_lane_id: str | None = None,
+    registry: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     from .capabilities.machine_configuration.builtins import (
         build_builtin_machine_configuration_registry,
         project_goal_with_builtin_machine_configuration,
     )
     from .capabilities.machine_configuration.store import read_machine_configuration
 
-    registry = load_registry(registry_path)
+    if registry is None:
+        registry = load_registry(registry_path)
     machine_configuration = read_machine_configuration(
         runtime_root,
         registry=build_builtin_machine_configuration_registry(),
@@ -515,6 +517,7 @@ def collect_status_history(
     status_include_runtime_goals: bool,
     activation_state_filter: GoalActivationState | str | None = None,
     agent_lane_id: str | None = None,
+    registry: dict[str, Any] | None = None,
 ) -> StatusHistoryCollection:
     history = collect_history(
         registry_path=registry_path,
@@ -524,6 +527,7 @@ def collect_status_history(
         include_runtime_goals=True,
         activation_state_filter=activation_state_filter,
         agent_lane_id=agent_lane_id,
+        registry=registry,
     )
     audit = build_run_history_audit(
         history,

loopx/promotion_gate.py

@@ -143,8 +143,10 @@ def build_promotion_gate(
     *,
     registry_path: Path,
     runtime_root_override: str | None,
+    registry: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
-    registry = load_registry(registry_path)
+    if registry is None:
+        registry = load_registry(registry_path)
     runtime_root = resolve_runtime_root(
         registry,
         runtime_root_override,