From 4fa61cc7b0e584665b859e2a801af076f37471c8 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sun, 1 Nov 2026 13:48:33 +0800 Subject: [PATCH] refactor(control-plane): share projection snapshot fallback Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/control_plane/effect_runtime.py | 13 +++++++++++++ loopx/control_plane/quota/turn_envelope.py | 12 ++---------- loopx/control_plane/turn_driver/loop_controller.py | 11 ++--------- tests/control_plane/test_context_delivery_cli.py | 6 +++--- .../test_progressive_native_context.py | 6 +++--- tests/control_plane/test_saved_context_project.py | 2 +- tests/test_loop_turn_loop_controller.py | 3 +-- tests/test_turn_controller_projection.py | 4 ++-- 8 files changed, 27 insertions(+), 30 deletions(-) diff --git a/loopx/control_plane/effect_runtime.py b/loopx/control_plane/effect_runtime.py index 76aa51b9d2..819088102a 100644 --- a/loopx/control_plane/effect_runtime.py +++ b/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.""" diff --git a/loopx/control_plane/quota/turn_envelope.py b/loopx/control_plane/quota/turn_envelope.py index 5d19712679..5011e7ce31 100644 --- a/loopx/control_plane/quota/turn_envelope.py +++ b/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) diff --git a/loopx/control_plane/turn_driver/loop_controller.py b/loopx/control_plane/turn_driver/loop_controller.py index 33cd1220d1..0e5ad27e49 100644 --- a/loopx/control_plane/turn_driver/loop_controller.py +++ b/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): diff --git a/tests/control_plane/test_context_delivery_cli.py b/tests/control_plane/test_context_delivery_cli.py index 5974effe93..ea9ea71955 100644 --- a/tests/control_plane/test_context_delivery_cli.py +++ b/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() diff --git a/tests/control_plane/test_progressive_native_context.py b/tests/control_plane/test_progressive_native_context.py index 2920beffc9..aab342d55f 100644 --- a/tests/control_plane/test_progressive_native_context.py +++ b/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"): diff --git a/tests/control_plane/test_saved_context_project.py b/tests/control_plane/test_saved_context_project.py index e1ac717efb..a451140ecf 100644 --- a/tests/control_plane/test_saved_context_project.py +++ b/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" ) diff --git a/tests/test_loop_turn_loop_controller.py b/tests/test_loop_turn_loop_controller.py index cb13529427..acc4e592f5 100644 --- a/tests/test_loop_turn_loop_controller.py +++ b/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 diff --git a/tests/test_turn_controller_projection.py b/tests/test_turn_controller_projection.py index 2e42fb6ff4..955ebfdb29 100644 --- a/tests/test_turn_controller_projection.py +++ b/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