refactor(control-plane): share projection snapshot fallback

loopx/control_plane/effect_runtime.py

@@ -1584,6 +1584,19 @@ def effect_runtime_result(
     ).get("result")
 
 
+def effect_runtime_projection_result(method: str, params: Mapping[str, Any]) -> Any:
+    """Keep projections inline unless a size refusal requires a local snapshot."""
+
+    try:
+        return effect_runtime_result(method, params)
+    except EffectRuntimeRejected as error:
+        if error.diagnostic_code != "request_too_large":
+            raise
+        # Reuse the existing bounded transport after refusal, never replay an
+        # ambiguous response or truncate the projection's complete source.
+        return effect_runtime_result(method, params, large_local_snapshot=True)
+
+
 def _serving_runtime_identity() -> dict[str, Any] | None:
     """Project the identity of the runtime serving this revision, if any."""
 

loopx/control_plane/quota/turn_envelope.py

@@ -7,7 +7,7 @@
 from dataclasses import dataclass
 from typing import Any
 
-from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result, source_number_tokens
+from ..effect_runtime import effect_runtime_projection_result, source_number_tokens
 from ..scheduler.execution_context import (
     SchedulerExecutionContextResolution,
     render_scheduler_execution_args,
@@ -60,15 +60,7 @@ def _turn_envelope_result(
         tokens = source_number_tokens(params)
         if tokens:
             request["source_number_tokens"] = tokens
-    try:
-        result = effect_runtime_result("quota.turn_envelope.evaluate", request)
-    except EffectRuntimeRejected as error:
-        if error.diagnostic_code != "request_too_large":
-            raise
-        # The client refused before sending this pure projection. Reuse the
-        # existing bounded local-snapshot transport, never truncate the source
-        # or re-enter quota admission. Small packets retain their normal path.
-        result = effect_runtime_result("quota.turn_envelope.evaluate", request, large_local_snapshot=True)
+    result = effect_runtime_projection_result("quota.turn_envelope.evaluate", request)
     if not isinstance(result, Mapping):
         raise RuntimeError("TypeScript Turn envelope result must be an object")
     return dict(result)

loopx/control_plane/turn_driver/loop_controller.py

@@ -10,7 +10,7 @@
 from typing import Any
 
 from ..effect_program import SettlementIdentity
-from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result
+from ..effect_runtime import effect_runtime_projection_result
 from ..quota.turn_envelope import TurnControllerInput
 from .driver import _historical_route_text
 from .transaction import LoopXTurnResultKind
@@ -42,14 +42,7 @@ def _controller_result(operation: str, **facts: Any) -> dict[str, Any]:
             key: str(value) if type(value) is int else None for key, value in scalars.items()
         }
     request = {"operation": operation, **facts}
-    try:
-        payload = effect_runtime_result("turn.controller", request)
-    except EffectRuntimeRejected as error:
-        if error.diagnostic_code != "request_too_large":
-            raise
-        # The pure request was refused before sending. Preserve complete facts
-        # through the existing bounded snapshot transport, as the envelope does.
-        payload = effect_runtime_result("turn.controller", request, large_local_snapshot=True)
+    payload = effect_runtime_projection_result("turn.controller", request)
     if not isinstance(payload, dict):
         raise RuntimeError("TypeScript Turn controller returned an invalid response")
     if payload.get("ok") is False and isinstance(payload.get("message"), str):

tests/control_plane/test_context_delivery_cli.py

@@ -883,7 +883,7 @@ def test_context_nonfinite_numbers_are_rejected_before_rpc(monkeypatch, number):
     def unexpected(*args, **kwargs):
         pytest.fail("Non-finite evidence must not enter the codec")
 
-    monkeypatch.setattr(turn_envelope, "effect_runtime_result", unexpected)
+    monkeypatch.setattr(turn_envelope, "effect_runtime_projection_result", unexpected)
     with pytest.raises(ValueError, match="finite"):
         prepare_context_delivery({"nested": [{"measurement": number}]},
             scope={}, capture_id="invalid")
@@ -894,7 +894,7 @@ def test_fractional_observations_keep_complete_codec_cost_proportional(tmp_path,
     from loopx.control_plane.quota import turn_envelope
 
     isolate_sqlite_runtime(tmp_path, monkeypatch)
-    original_rpc = turn_envelope.effect_runtime_result
+    original_rpc = turn_envelope.effect_runtime_projection_result
     costs = []
 
     def size(value):
@@ -906,7 +906,7 @@ def measured_rpc(method, params, **kwargs):
         costs.append(size(params) + size(result))
         return result
 
-    monkeypatch.setattr(turn_envelope, "effect_runtime_result", measured_rpc)
+    monkeypatch.setattr(turn_envelope, "effect_runtime_projection_result", measured_rpc)
     totals, source_sizes = [], []
     for fraction in (False, True):
         costs.clear()

tests/control_plane/test_progressive_native_context.py

@@ -308,7 +308,7 @@ def unavailable(*args, **kwargs):
         raise EffectRuntimeStartupError("Synthetic unavailable runtime", diagnostic_code="fixture_unavailable")
 
     setup(tmp_path, monkeypatch)
-    monkeypatch.setattr(turn_envelope, "effect_runtime_result", unavailable)
+    monkeypatch.setattr(turn_envelope, "effect_runtime_projection_result", unavailable)
     original = payload()
     _, required = project(tmp_path, payload=original)
     recovery = context._state(tmp_path)["recovery"]
@@ -346,14 +346,14 @@ def test_native_required_view_and_encoding_use_one_runtime_request(tmp_path, mon
     from loopx.control_plane.quota import turn_envelope
 
     setup(tmp_path, monkeypatch)
-    invoke = turn_envelope.effect_runtime_result
+    invoke = turn_envelope.effect_runtime_projection_result
     requests = []
 
     def observe(method, params, **kwargs):
         requests.append((method, params["operation"]))
         return invoke(method, params, **kwargs)
 
-    monkeypatch.setattr(turn_envelope, "effect_runtime_result", observe)
+    monkeypatch.setattr(turn_envelope, "effect_runtime_projection_result", observe)
     original = payload()
     before = json.dumps(original, sort_keys=True)
     for capture in ("cold", "warm"):

tests/control_plane/test_saved_context_project.py

@@ -274,7 +274,7 @@ def test_saved_projection_planner_process_exit_keeps_current_recovery(tmp_path,
         "import os\n"
         "from loopx.control_plane.quota import turn_envelope\n"
         "def interrupt(*args, **kwargs): os._exit(37)\n"
-        "turn_envelope.effect_runtime_result = interrupt\n"
+        "turn_envelope.effect_runtime_projection_result = interrupt\n"
         "from loopx.cli import main\n"
         "raise SystemExit(main())\n"
     )

tests/test_loop_turn_loop_controller.py

@@ -398,7 +398,6 @@ def test_controller_snapshot_limit_rejects_before_sending(monkeypatch, qualified
 @pytest.mark.parametrize("rejection", ["invalid_request", "ambiguous"])
 def test_controller_does_not_retry_other_transport_failures(monkeypatch, rejection) -> None:
     from loopx.control_plane import effect_runtime
-    from loopx.control_plane.turn_driver import loop_controller
 
     failure = (effect_runtime.EffectRuntimeResponseAmbiguous("turn.controller", timeout=1)
                if rejection == "ambiguous" else effect_runtime.EffectRuntimeRejected("rejected"))
@@ -408,7 +407,7 @@ def reject(method, params, **kwargs):
         calls.append((method, kwargs))
         raise failure
 
-    monkeypatch.setattr(loop_controller, "effect_runtime_result", reject)
+    monkeypatch.setattr(effect_runtime, "effect_runtime_result", reject)
     with pytest.raises(type(failure)) as error:
         _decide_loop_disposition(turn_receipt=None, quota_decision=_envelope(should_run=True))
     assert error.value is failure

tests/test_turn_controller_projection.py

@@ -58,13 +58,13 @@ def record(operation, **params):
     assert expected["disposition"] == ("user_action_required" if user_action_required else "wait")
     calls.clear()
     controller_calls = []
-    original_runtime = loop_controller.effect_runtime_result
+    original_runtime = loop_controller.effect_runtime_projection_result
 
     def record_runtime(method, params, **options):
         controller_calls.append((method, params["operation"]))
         return original_runtime(method, params, **options)
 
-    monkeypatch.setattr(loop_controller, "effect_runtime_result", record_runtime)
+    monkeypatch.setattr(loop_controller, "effect_runtime_projection_result", record_runtime)
     actual = decide_managed_step(journal, current, goal_id=GOAL_ID,
         agent_id=AGENT_ID, turn_key=journal["turn_key"])
     assert actual == expected