test(cli-output): declare the bounded managed executor readback budget

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>

Author: huangruiteng

examples/control_plane/cli-output-probe-runner.py

@@ -108,6 +108,9 @@ def _receipt_row(
         "reward_memory_outcome_prompt_revision": (
             semantics.reward_memory_outcome_prompt_revision(text)
         ),
+        "managed_executor_binding_revision": (
+            semantics.managed_executor_binding_revision(text)
+        ),
         "guided_todo_delta_schema_versions": (
             semantics.guided_todo_delta_schema_versions(payload)
             if isinstance(payload, dict)

loopx/control_plane/testing/cli_output_differential.py

@@ -177,6 +177,45 @@ class GrowthAllowance:
     "compact_payload_chars": 640,
 }
 
+# The Turn plan readback adds one bounded managed-executor binding so a caller
+# sees which executor a planned Turn would use and whether it can launch here,
+# instead of inferring it from the host id.  The allowance is bound to the
+# declared none-to-v0 binding transition and to the Turn surfaces that quote
+# it; quota, status, and every other agent-facing surface keep the ordinary
+# budget, and once v0 is the baseline a v0-to-v0 change receives no allowance.
+_MANAGED_EXECUTOR_BINDING_V0_MIGRATION_GROWTH_ALLOWANCE: dict[Metric, int] = {
+    "chars": 512,
+    "utf8_bytes": 512,
+    "lines": 12,
+    "compact_payload_chars": 448,
+}
+
+_MANAGED_EXECUTOR_BINDING_SURFACES = frozenset(
+    {
+        "loopx_turn_plan",
+        "loopx_turn_plan_transaction_detail",
+        "loopx_turn_run_once_preview",
+    }
+)
+
+
+def _managed_executor_binding_allowance(
+    row_id: str,
+    base: Mapping[str, Any],
+    candidate: Mapping[str, Any],
+    metric: Metric,
+) -> int:
+    surface = row_id.partition("/")[2].partition("/")[0]
+    if (
+        row_id.startswith(("surface/", "variant/"))
+        and surface in _MANAGED_EXECUTOR_BINDING_SURFACES
+        and base.get("managed_executor_binding_revision") is None
+        and candidate.get("managed_executor_binding_revision")
+        == "managed_executor_binding_v0"
+    ):
+        return _MANAGED_EXECUTOR_BINDING_V0_MIGRATION_GROWTH_ALLOWANCE[metric]
+    return 0
+
 
 def _reward_memory_outcome_prompt_allowance(
     row_id: str,
@@ -545,6 +584,12 @@ def _compare_row(base: dict[str, Any], candidate: dict[str, Any]) -> dict[str, A
                 candidate,
                 metric,
             ),
+            _managed_executor_binding_allowance(
+                row_id,
+                base,
+                candidate,
+                metric,
+            ),
         )
         # Thin installed prompts contain bilingual lifecycle instructions. A
         # small character-level clarification can cost three bytes per CJK

loopx/control_plane/testing/cli_output_semantics.py

@@ -41,6 +41,28 @@ def reward_memory_outcome_prompt_revision(text: str) -> str | None:
         else None
     )
 
+
+def managed_executor_binding_revision(text: str) -> str | None:
+    """Attribute the managed-executor binding readback on a Turn surface.
+
+    This is qualification evidence for the exact projection, never a runtime
+    classifier: the binding key alone would match prose, so the revision also
+    requires the executor identity, its launchability claim, and the typed
+    reason slot that only this readback renders.
+    """
+
+    required = (
+        '"managed_executor"',
+        '"executor_kind"',
+        '"available"',
+        '"unavailable_reason"',
+    )
+    return (
+        "managed_executor_binding_v0"
+        if all(fragment in text for fragment in required)
+        else None
+    )
+
 _MARKDOWN_HEADING = re.compile(r"^#{1,6}\s+.+$")
 _RUNTIME_ROOT_COMMAND_ROUTE = re.compile(
     r"(?m)(?:^|[\"'`])[^\r\n\S]*loopx\s+--runtime-root\s+"

tests/control_plane/test_cli_output_differential.py

@@ -146,6 +146,51 @@ def test_reward_memory_outcome_prompt_budget_is_one_time_bounded_and_prompt_only
     ]
 
 
+def test_managed_executor_binding_budget_is_one_time_bounded_and_turn_only() -> None:
+    from loopx.control_plane.testing.cli_output_differential import _compare_row
+    from loopx.control_plane.testing.cli_output_semantics import (
+        managed_executor_binding_revision,
+    )
+
+    projection = (
+        '{\n  "managed_executor": {\n'
+        '    "executor_kind": "managed",\n'
+        '    "available": true,\n'
+        '    "unavailable_reason": null\n  }\n}'
+    )
+    assert (
+        managed_executor_binding_revision(projection)
+        == "managed_executor_binding_v0"
+    )
+    assert (
+        managed_executor_binding_revision(
+            projection.replace('"unavailable_reason"', '"reason"')
+        )
+        is None
+    )
+
+    base = _row(row_id="variant/loopx_turn_run_once_preview/small/json")
+    current = {
+        **base,
+        "chars": base["chars"] + 254,
+        "utf8_bytes": base["utf8_bytes"] + 254,
+        "lines": base["lines"] + 9,
+        "compact_payload_chars": base["compact_payload_chars"] + 205,
+        "managed_executor_binding_revision": "managed_executor_binding_v0",
+    }
+    assert not _compare_row(base, current)["failures"]
+    assert _compare_row(base, {**current, "chars": base["chars"] + 513})[
+        "failures"
+    ]
+    assert _compare_row(current, {**current, "chars": current["chars"] + 254})[
+        "failures"
+    ]
+    other = {**base, "row_id": "surface/status/small/json"}
+    assert _compare_row(other, {**current, "row_id": other["row_id"]})[
+        "failures"
+    ]
+
+
 def test_regular_integration_pr_keeps_requested_cli_output_base() -> None:
     ancestors = {
         ("origin/main", "HEAD"),