Reuse identical CLI settlement plans within a quota decision

loopx/control_plane/quota/effect_program.py

@@ -2,10 +2,13 @@
 
 import os
 import shlex
-from collections.abc import Mapping
+from collections.abc import Iterator, Mapping
+from contextlib import contextmanager
+from contextvars import ContextVar
+from copy import deepcopy
 from typing import Any
 
-from ..effect_runtime import effect_runtime_result
+from ..effect_runtime import _encode_rpc_json, effect_runtime_result
 from ..effect_program import (
     SETTLEMENT_IDENTITY_SCHEMA_VERSION,
     SETTLEMENT_PLAN_SCHEMA_VERSION,
@@ -33,6 +36,34 @@
 
 _PLATFORM_IS_POSIX = os.name == "posix"
 
+_CLI_PLAN_PROJECTIONS: ContextVar[dict[bytes, Any] | None] = ContextVar(
+    "quota_cli_plan_projections",
+    default=None,
+)
+
+
+@contextmanager
+def reuse_cli_settlement_plans() -> Iterator[None]:
+    """Reuse pure rendering within one quota decision's projection passes."""
+    token = _CLI_PLAN_PROJECTIONS.set({})
+    try:
+        yield
+    finally:
+        _CLI_PLAN_PROJECTIONS.reset(token)
+
+
+def _render_cli_settlement_plan(request: dict[str, Any]) -> Any:
+    projections = _CLI_PLAN_PROJECTIONS.get()
+    if projections is None:
+        return effect_runtime_result("settlement.turn_scoped_cli_plan", request)
+    # Key by the complete typed request, including scope and command
+    # arguments. Each projection gets an independent copy to edit.
+    key = _encode_rpc_json(request)
+    if key not in projections:
+        projections[key] = effect_runtime_result("settlement.turn_scoped_cli_plan", request)
+    return deepcopy(projections[key])
+
+
 __all__ = [
     "SETTLEMENT_IDENTITY_SCHEMA_VERSION",
     "SETTLEMENT_PLAN_SCHEMA_VERSION",
@@ -281,7 +312,7 @@ def writeback_command(boundary: str) -> str:
         f"--source {quota_spend_source} --execute{binding_arg}{turn_arg}{goal_ref_arg}"
         f"{scoped_cli_args}{max_output_arg}"
     )
-    payload = effect_runtime_result("settlement.turn_scoped_cli_plan", {
+    payload = _render_cli_settlement_plan({
         "identity": identity.as_dict(), "delivery_boundary": delivery_boundary,
         "autonomous_replan": autonomous_replan,
         "handoff_mode": handoff_mode,

loopx/control_plane/quota/should_run_packet.py

@@ -4,7 +4,7 @@
 import json
 from .effective_action import EffectiveAction
 from ..effect_runtime import effect_runtime_result
-from .effect_program import ReceiptBoundReplayPhase
+from .effect_program import ReceiptBoundReplayPhase, reuse_cli_settlement_plans
 from collections.abc import Callable, Mapping
 from dataclasses import dataclass
 from pathlib import Path
@@ -1671,6 +1671,7 @@ def _build_active_quota_payload(
     return payload
 
 
+@reuse_cli_settlement_plans()
 def _build_quota_should_run_payload(
     prepared: _QuotaDecisionPreparation,
     route: _QuotaDecisionRoute,

tests/control_plane/test_quota_settlement.py

@@ -81,6 +81,47 @@ def test_writeback_guidance_uses_current_cli_scale_vocabulary():
         assert alias not in hint  # Historical aliases remain read-only.
 
 
+def test_quota_decision_reuses_only_the_same_settlement_plan_input(monkeypatch):
+    real_render = quota_effect_program.effect_runtime_result
+    requests = []
+
+    def record(method, request):
+        assert method == "settlement.turn_scoped_cli_plan"
+        requests.append(request)
+        return real_render(method, request)
+
+    monkeypatch.setattr(quota_effect_program, "effect_runtime_result", record)
+
+    def plan(capabilities=""):
+        return quota_effect_program.build_turn_scoped_cli_settlement_plan(
+            goal_id=GOAL_ID,
+            agent_id=AGENT_ID,
+            todo_id=TODO_ID,
+            turn_instance_id=TURN_ID,
+            scoped_cli_args=capabilities,
+            lifecycle_actor_args="",
+        )
+
+    with quota_effect_program.reuse_cli_settlement_plans():
+        first = plan()
+        expected = first.as_dict()
+        writeback = next(step for step in first.steps if step.kind is SettlementStepKind.DURABLE_WRITEBACK)
+        assert isinstance(writeback.vision_authoring, dict)
+        writeback.vision_authoring.clear()
+        assert plan().as_dict() == expected
+        assert len(requests) == 1
+
+        with_capability = settlement_step_command(
+            plan(" --available-capability shell").as_dict(),
+            SettlementStepKind.QUOTA_SPEND,
+        )
+        assert "--available-capability shell" in with_capability
+        assert len(requests) == 2
+
+    assert plan().as_dict() == expected
+    assert len(requests) == 3
+
+
 @pytest.mark.parametrize("replan", [False, True])
 def test_command_plan_supplies_actor_from_identity(replan):
     plan = quota_effect_program.build_turn_scoped_cli_settlement_plan(