From 73d10d1c5d7163755397f8925c3fd0a3e7983f6e Mon Sep 17 00:00:00 2001
From: huangruiteng <huangrt01@163.com>
Date: Wed, 4 Nov 2026 22:26:48 +0800
Subject: [PATCH] refactor(quota): keep execution preparation presentation with
 its owner

Signed-off-by: huangruiteng <huangrt01@163.com>
---
 .../quota/execution_preparation.py            | 31 +++++++++++++++++
 .../work_items/interaction_contract.py        | 34 ++-----------------
 2 files changed, 33 insertions(+), 32 deletions(-)

diff --git a/loopx/control_plane/quota/execution_preparation.py b/loopx/control_plane/quota/execution_preparation.py
index e379d02b9d..792c508937 100644
--- a/loopx/control_plane/quota/execution_preparation.py
+++ b/loopx/control_plane/quota/execution_preparation.py
@@ -156,3 +156,34 @@ def apply_selected_todo_execution_preparation(
         payload["actionable_by_codex"] = False
         if obligation is not None:
             obligation["must_attempt_work"] = False
+
+
+def attach_interaction_execution_preparation(
+    contract: dict[str, Any], payload: dict[str, Any],
+) -> None:
+    preparation = payload.get("execution_preparation")
+    if not isinstance(preparation, dict):
+        return
+    contract["agent_channel"]["execution_preparation"] = preparation
+    contract["cli_channel"]["execution_preparation"] = preparation
+    state = preparation.get("state")
+    if state == "ready":
+        return
+    reason = str(preparation.get("reason") or "Selected Todo execution is not prepared.")
+    contract["agent_channel"]["delivery_allowed"] = False
+    contract["agent_channel"]["primary_action"] = reason
+    if state != "claim_required":
+        return
+    claim = preparation.get("claim")
+    command = claim.get("command_line") if isinstance(claim, dict) else None
+    if isinstance(command, str) and command:
+        contract["agent_channel"]["must_attempt"] = True
+        contract["agent_channel"]["quiet_noop_allowed"] = False
+        contract["agent_channel"]["primary_action"] = (
+            f"Run the atomic execution preparation action: {command}"
+        )
+        contract["cli_channel"]["next_cli_actions"] = [command]
+        for key in (
+            "settlement_plan", "quota_spend_source", "replan_settlement_contract",
+        ):
+            contract["cli_channel"].pop(key, None)
diff --git a/loopx/control_plane/work_items/interaction_contract.py b/loopx/control_plane/work_items/interaction_contract.py
index bf63a0bb60..dad4f6819b 100644
--- a/loopx/control_plane/work_items/interaction_contract.py
+++ b/loopx/control_plane/work_items/interaction_contract.py
@@ -3,6 +3,7 @@
 from ..goals.acceptance_readback import configured_checks
 from ..quota.blocked_transition_notice import blocked_priority_fallback_owner_reason
 from ..quota.effective_action import EffectiveAction
+from ..quota.execution_preparation import attach_interaction_execution_preparation
 from ..effect_runtime import effect_runtime_result
 import json
 import shlex
@@ -1578,37 +1579,6 @@ def _attach_interaction_vision_wait_state(
     }
 
 
-def _attach_interaction_execution_preparation(
-    contract: dict[str, Any], payload: dict[str, Any],
-) -> None:
-    preparation = payload.get("execution_preparation")
-    if not isinstance(preparation, dict):
-        return
-    contract["agent_channel"]["execution_preparation"] = preparation
-    contract["cli_channel"]["execution_preparation"] = preparation
-    state = preparation.get("state")
-    if state == "ready":
-        return
-    reason = str(preparation.get("reason") or "Selected Todo execution is not prepared.")
-    contract["agent_channel"]["delivery_allowed"] = False
-    contract["agent_channel"]["primary_action"] = reason
-    if state != "claim_required":
-        return
-    claim = preparation.get("claim")
-    command = claim.get("command_line") if isinstance(claim, dict) else None
-    if isinstance(command, str) and command:
-        contract["agent_channel"]["must_attempt"] = True
-        contract["agent_channel"]["quiet_noop_allowed"] = False
-        contract["agent_channel"]["primary_action"] = (
-            f"Run the atomic execution preparation action: {command}"
-        )
-        contract["cli_channel"]["next_cli_actions"] = [command]
-        for key in (
-            "settlement_plan", "quota_spend_source", "replan_settlement_contract",
-        ):
-            contract["cli_channel"].pop(key, None)
-
-
 def _interaction_fallback_policy_required(payload: Mapping[str, object], *, mode: str) -> bool:
     return mode in {
         "user_gate",
@@ -1747,7 +1717,7 @@ def _build_interaction_contract_projection(
             registry_path=registry_path,
         ),
     }
-    _attach_interaction_execution_preparation(contract, payload)
+    attach_interaction_execution_preparation(contract, payload)
     response_plan = _build_interaction_response_plan(
         user_channel=user_channel,
         agent_channel=agent_channel,