diff --git a/docs/reference/local-delegation.md b/docs/reference/local-delegation.md
index c4c51567b8..22676fd28b 100644
--- a/docs/reference/local-delegation.md
+++ b/docs/reference/local-delegation.md
@@ -159,6 +159,32 @@ delegate read --operation-id review-round-1
 delegate wait --operation-id review-round-1
 ```
 
+Inspection uses the bound worker workspace as its actual safety scan root. If
+quota defers the exact Todo for control repair, inspection returns
+`state: turn_blocked` with `turn_blocker.reason_code`, the original selection
+state and a contract-error count. Canonical acceptance can still be ready;
+`executor: null` means it was not inspected, not that the model runtime failed.
+Use `loopx check --scan-root /absolute/reviewer-worktree` with the same registry
+to diagnose the scan. Repair the source or configuration, then inspect again.
+Do not exempt tests, scan the installed package instead, retarget the Todo or
+start another operation to bypass the refusal. Other unstructured CLI failures
+remain errors, and a refusal whose bounded fields are malformed — including a
+`state` that is not one of the decoded string literals — fails closed instead of
+reporting `turn_blocked`. The observation starts no host, Turn journal or quota spend.
+Normal quota selection may still admit unrelated eligible work; this preflight
+never substitutes another Todo.
+
+中文：预检以 binding 固定的真实 worker 工作树作为安全扫描根。quota 因控制面
+修复延后该精确 Todo 时，返回 `state: turn_blocked`、原选路状态、
+`turn_blocker.reason_code` 和契约错误数；规范验收可能仍已就绪。
+`executor: null` 表示未检查执行器，不表示模型故障。使用相同 registry 和
+`loopx check --scan-root /absolute/reviewer-worktree` 定位，再修复原来源或配置
+并重做预检。不能豁免测试目录、改扫安装包、换 Todo 或创建新操作绕过拒绝。
+其他无结构 CLI 故障仍报错；拒绝投影字段畸形（含 `state` 不是已解码字符串
+字面量）时按失败关闭报错，不返回 `turn_blocked`；该观察不启动 host、Turn
+journal 或扣额。
+普通 quota 选路仍可安排其他独立且合格的工作；本预检不会替换 Todo。
+
 `request.json` contains the same `collaboration_brief_v0` used by MCP:
 
 ```json
diff --git a/loopx/cli_commands/turn_decision.py b/loopx/cli_commands/turn_decision.py
index f93667b0d2..8db3b1f10b 100644
--- a/loopx/cli_commands/turn_decision.py
+++ b/loopx/cli_commands/turn_decision.py
@@ -35,6 +35,7 @@
     scheduler_execution_context_for_turn,
 )
 from ..status import AUTONOMOUS_REPLAN_PERIODIC_LOOKBACK, collect_status
+from ..control_plane.effect_runtime import effect_runtime_result
 from .lark_inbox import build_lark_operator_inbox_urgency_projector
 from .turn_selection import turn_controller_advisory_primary
 
@@ -42,6 +43,19 @@
 TURN_DECISION_ROUTE_SOURCE = "loopx_turn_plan"
 
 
+class TurnTodoSelectionError(ValueError):
+    """Transport quota's existing refusal, without a new admission rule."""
+
+    def __init__(self, projection: dict[str, Any]) -> None:
+        self.code = projection["error_code"]
+        self.payload = {"selection_rejection": projection["selection_rejection"]}
+        reason = projection["selection_rejection"].get("reason_code")
+        super().__init__(
+            "Requested Turn Todo is not currently eligible; no alternate task was selected"
+            + (f"; reason={reason}" if reason else "")
+        )
+
+
 def collect_turn_status_payload(
     args: argparse.Namespace,
     *,
@@ -164,7 +178,11 @@ def resolve(self) -> dict[str, Any]:
         decision = self.build_turn_decision(requested_action_todo_id=self.requested_todo_id)
         selected = decision.get("selected_todo")
         if not isinstance(selected, dict) or selected.get("todo_id") != self.requested_todo_id:
-            raise ValueError("Requested Turn Todo is not currently eligible; no alternate task was selected")
+            summary = self.status_payload.get("contract_summary")
+            raise TurnTodoSelectionError(effect_runtime_result("turn.selection.rejection", {
+                "requested_todo_id": self.requested_todo_id, "decision": decision,
+                "contract_error_count": summary.get("errors") if isinstance(summary, Mapping) else None,
+            }))
         selected["selected_by"] = "turn_explicit_todo"
         return decision
 
diff --git a/loopx/collaboration_mcp.py b/loopx/collaboration_mcp.py
index 4c2ef7346e..c38ab1486e 100644
--- a/loopx/collaboration_mcp.py
+++ b/loopx/collaboration_mcp.py
@@ -356,7 +356,7 @@ def inspect(self, binding_id: str) -> dict:
                 or selected_scan_root.resolve() != workspace):
             raise ValueError("delegation inspection cannot execute or retarget bound work")
         preview = self._cli(binding, *arguments)
-        if preview.get("status") != "preview":
+        if preview.get("status") != "preview" and "selection_rejection" not in preview:
             raise ValueError(f"delegation Turn preflight unavailable: {preview.get('error') or preview.get('status')}")
         current = delegation_validation.capture(self, binding)
         if acceptance != current or self.binding(binding_id, require_active=True) != binding:
diff --git a/loopx/control_plane/collaboration/delegation.ts b/loopx/control_plane/collaboration/delegation.ts
index be8ffe982b..e8d610a740 100644
--- a/loopx/control_plane/collaboration/delegation.ts
+++ b/loopx/control_plane/collaboration/delegation.ts
@@ -6,6 +6,7 @@ import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts";
 import {canonicalAuthoritySha256} from "../coordination/authority_store_codec.ts";
 import {acceptanceValidationEffects, type AcceptanceCompletionRequirements} from "../goals/acceptance_contract.ts";
 import {normalizeTodoCompletionValidationDeclaration} from "../todos/completion_validation_declaration.ts";
+import {readTurnSelectionRejection, turnSelectionRejectionState} from "../turn_driver/selection_rejection.ts";
 
 function requireThat(ok: unknown, message: string): asserts ok {
   if (!ok) throw new EffectRuntimeRequestError(message);
@@ -170,6 +171,28 @@ export function delegationPreflight(params: JsonObject): JsonObject {
   }
   const preview = requireJsonObject(params.preview, "Turn preview");
   const effects = requireJsonObject(preview.effects, "preview effects");
+  if (preview.ok === false && preview.selection_rejection !== undefined) {
+    const refusal = readTurnSelectionRejection(preview.selection_rejection, binding.todo_id);
+    const refusalState = turnSelectionRejectionState(refusal.state);
+    requireThat(preview.effects_scope === "current_invocation"
+      && ["host_invoked", "state_written", "quota_spent", "scheduler_acknowledged"].every(k => effects[k] === false)
+      && refusal.schema_version === "loopx_turn_selection_rejection_v0"
+      && refusal.source === "quota.should-run" && refusal.requested_todo_id === binding.todo_id
+      && refusalState !== null
+      && preview.error_code === `turn_todo_selection_${refusalState}`,
+    "delegation inspection requires a matching effect-free selection refusal");
+    const acceptance = params.acceptance === null ? null : requireJsonObject(params.acceptance, "task acceptance");
+    return {
+      schema_version: "loopx_delegation_preflight_v0", binding, state: "turn_blocked",
+      turn_eligible: false, turn_route: null, turn_blocker: refusal,
+      acceptance_ready: acceptance?.todo_id === binding.todo_id && acceptance?.state === "ready"
+        && params.validation_files_current === true,
+      authority_ready: true, authority_reason: null, authority_state: "promoted",
+      authority_next_action: "none", promotion_from_surface_allowed: false,
+      executor: null, effects,
+      note: "Quota refused this exact Todo before host or executor inspection. Read status/check with the bound workspace scan root; do not retarget this inspection or bypass repair.",
+    };
+  }
   requireThat(preview.dry_run === true && preview.status === "preview"
     && ["host_invoked", "state_written", "quota_spent", "scheduler_acknowledged"].every(k => effects[k] === false),
   "delegation inspection requires a read-only Turn preview");
diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts
index bdc5be9b97..4ecc815623 100644
--- a/loopx/control_plane/effect_runtime_handlers.ts
+++ b/loopx/control_plane/effect_runtime_handlers.ts
@@ -89,6 +89,7 @@ import {
   type TurnJournalInspectionRequest,
 } from "./turn_driver/turn_journal.ts";
 import { commitTurnJournal } from "./turn_driver/turn_journal_effects.ts";
+import { projectTurnSelectionRejection } from "./turn_driver/selection_rejection.ts";
 import {
   evaluateTodoCompletionFence,
 } from "./todos/completion_fence.ts";
@@ -714,6 +715,7 @@ export function createEffectRuntimeHandlers(
       evaluatePostWritebackHookTransaction,
     ],
     ["collaboration.delegation.binding", selectDelegationBinding],
+    ["turn.selection.rejection", projectTurnSelectionRejection],
     ["collaboration.delegation.preflight", delegationPreflight],
     ["collaboration.delegation.validation_plan", delegationValidationPlan],
     ["collaboration.delegation.turn_plan", delegationTurnPlanDecision],
diff --git a/loopx/control_plane/turn_driver/selection_rejection.ts b/loopx/control_plane/turn_driver/selection_rejection.ts
new file mode 100644
index 0000000000..f796e3fdce
--- /dev/null
+++ b/loopx/control_plane/turn_driver/selection_rejection.ts
@@ -0,0 +1,60 @@
+/** Observe a refused selection from quota's decision; never decide admission. */
+import type {JsonObject} from "../effect_program.ts";
+import {isStringLiteral, requireJsonObject} from "../runtime_decode.ts";
+import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts";
+
+function code(value: unknown): string | null {
+  return typeof value === "string" && /^[a-z][a-z0-9_]{0,159}$/.test(value) ? value : null;
+}
+
+const REJECTION_STATES = ["deferred", "rejected", "unavailable"] as const;
+type RejectionState = (typeof REJECTION_STATES)[number];
+
+/** Decode the state literal itself; a non-string (e.g. a JSON array) is not a refusal state. */
+export function turnSelectionRejectionState(value: unknown): RejectionState | null {
+  return typeof value === "string" && isStringLiteral(value, REJECTION_STATES) ? value : null;
+}
+
+export function readTurnSelectionRejection(value: unknown, requested: unknown): JsonObject {
+  const row = requireJsonObject(value, "Turn selection refusal");
+  const state = turnSelectionRejectionState(row.state);
+  if (row.schema_version !== "loopx_turn_selection_rejection_v0" || row.source !== "quota.should-run"
+    || row.requested_todo_id !== requested || state === null
+    || (row.reason_code !== null && code(row.reason_code) === null)
+    || (row.recovery_action !== null && code(row.recovery_action) === null)
+    || !Array.isArray(row.delivery_preemptions) || row.delivery_preemptions.length > 8
+    || row.delivery_preemptions.some(value => code(value) === null)
+    || ![true,false,null].includes(row.status_health_ok as boolean | null)
+    || (row.contract_error_count !== null && (!Number.isSafeInteger(row.contract_error_count) || Number(row.contract_error_count) < 0)))
+    throw new EffectRuntimeRequestError("matching bounded Turn selection refusal required");
+  return Object.fromEntries(["schema_version","source","requested_todo_id","state","reason_code",
+    "delivery_preemptions","recovery_action","status_health_ok","contract_error_count"]
+    .map(key => [key, key === "state" ? state : row[key]]));
+}
+
+export function projectTurnSelectionRejection(params: JsonObject): JsonObject {
+  const requested = params.requested_todo_id;
+  if (typeof requested !== "string" || requested.length < 1 || requested.length > 256)
+    throw new EffectRuntimeRequestError("bounded requested Todo identity required");
+  const decision = requireJsonObject(params.decision, "current quota decision");
+  const raw = decision.action_selection_qualification;
+  const qualification = raw && typeof raw === "object" && !Array.isArray(raw) ? raw as JsonObject : {};
+  const refused = qualification.requested_todo_id === requested
+    ? turnSelectionRejectionState(qualification.state) : null;
+  const state = refused === "deferred" || refused === "rejected" ? refused : "unavailable";
+  const reasons = Array.isArray(qualification.delivery_preemptions)
+    ? qualification.delivery_preemptions.filter(value => code(value) !== null).slice(0, 8) : [];
+  const count = params.contract_error_count;
+  return {
+    error_code: `turn_todo_selection_${state}`,
+    selection_rejection: {
+      schema_version: "loopx_turn_selection_rejection_v0", source: "quota.should-run",
+      requested_todo_id: requested, state,
+      reason_code: state === "unavailable" ? null : code(qualification.reason),
+      delivery_preemptions: state === "unavailable" ? [] : reasons,
+      recovery_action: state === "unavailable" ? null : code(qualification.recovery_action),
+      status_health_ok: typeof decision.status_health_ok === "boolean" ? decision.status_health_ok : null,
+      contract_error_count: Number.isSafeInteger(count) && Number(count) >= 0 ? count : null,
+    },
+  };
+}
diff --git a/tests/control_plane_ts/delegation.test.ts b/tests/control_plane_ts/delegation.test.ts
index eba606ec9c..cac137ae6d 100644
--- a/tests/control_plane_ts/delegation.test.ts
+++ b/tests/control_plane_ts/delegation.test.ts
@@ -2,6 +2,7 @@ import test from "node:test";
 import assert from "node:assert/strict";
 import {recordDelegationAdoption, delegationInventoryItem, delegationInventoryQuery, delegationPreflight, delegationTurnPlanDecision, delegationValidationPlan, recoverValidatedDelegationSettlement, selectDelegationBinding, transitionDelegationObservation} from "../../loopx/control_plane/collaboration/delegation.ts";
 import {canonicalAuthoritySha256} from "../../loopx/control_plane/coordination/authority_store_codec.ts";
+import {projectTurnSelectionRejection} from "../../loopx/control_plane/turn_driver/selection_rejection.ts";
 
 const binding = {id: "review", agent_id: "reviewer", todo_id: "todo_review", workspace: "/fixture",
   requesters: ["coordinator", "analyst"], host_args: ["--host", "dsh"], timeout_seconds: 60, output_refs: ["output.json"]};
@@ -15,6 +16,29 @@ const validationTodo = {todo_id: binding.todo_id, done: false, status: "open",
 const validationBasis = {status: "loaded", provider_revision: "fixture:1", todo: validationTodo,
   completion_requirements: null};
 
+test("preflight preserves a quota refusal, but rejects effectful or retargeted errors", () => {
+  const projected = projectTurnSelectionRejection({requested_todo_id:binding.todo_id,contract_error_count:3,
+    decision:{status_health_ok:false,action_selection_qualification:{state:"deferred",
+      requested_todo_id:binding.todo_id,reason:"control_repair",recovery_action:"reenter_guard_without_selection"}}});
+  const preview = {ok:false,...projected,effects_scope:"current_invocation",
+    effects:{host_invoked:false,state_written:false,quota_spent:false,scheduler_acknowledged:false}};
+  const input = {binding,preview,acceptance:{todo_id:binding.todo_id,state:"ready"},validation_files_current:true};
+  const observed = delegationPreflight(input);
+  assert.equal(observed.state,"turn_blocked"); assert.equal(observed.turn_eligible,false);
+  assert.equal(observed.acceptance_ready,true); assert.equal(observed.executor,null);
+  assert.deepEqual(observed.turn_blocker,projected.selection_rejection);
+  assert.deepEqual(delegationPreflight({...input,preview:{...preview,selection_rejection:{
+    ...(projected.selection_rejection as Record<string,unknown>),private_context:"not exported"}}}).turn_blocker,projected.selection_rejection);
+  for (const effects of [{...preview.effects,host_invoked:true}, {...preview.effects,state_written:null}])
+    assert.throws(() => delegationPreflight({...input,preview:{...preview,effects}}),/effect-free/);
+  assert.throws(() => delegationPreflight({...input,preview:{...preview,
+    selection_rejection:{...(projected.selection_rejection as Record<string,unknown>),requested_todo_id:"other"}}}),/matching/);
+  for (const raw of [["deferred"],["rejected"],["unavailable"]])
+    assert.throws(() => delegationPreflight({...input,preview:{...preview,
+      selection_rejection:{...(projected.selection_rejection as Record<string,unknown>),state:raw},
+      error_code:`turn_todo_selection_${raw[0]}`}}),/matching/);
+});
+
 test("independent delegation requires the current canonical declaration, not a Goal-wide contract", () => {
   const plan = delegationValidationPlan({binding, basis: validationBasis, declaration});
   assert.equal(plan.state, "ready");
diff --git a/tests/control_plane_ts/turn_selection_rejection.test.ts b/tests/control_plane_ts/turn_selection_rejection.test.ts
new file mode 100644
index 0000000000..c33eae57d2
--- /dev/null
+++ b/tests/control_plane_ts/turn_selection_rejection.test.ts
@@ -0,0 +1,57 @@
+import test from "node:test";
+import assert from "node:assert/strict";
+import {projectTurnSelectionRejection, readTurnSelectionRejection,
+  turnSelectionRejectionState} from "../../loopx/control_plane/turn_driver/selection_rejection.ts";
+
+const qualification = {schema_version:"action_selection_qualification_v0", state:"deferred",
+  requested_todo_id:"todo_worker", reason:"control_repair", recovery_action:"reenter_guard_without_selection",
+  delivery_preemptions:["control_repair","delivery_not_allowed"]};
+const params = {requested_todo_id:"todo_worker", contract_error_count:3,
+  decision:{status_health_ok:false, action_selection_qualification:qualification}};
+
+test("rejection reflects the quota reason and health count without another eligibility decision", () => {
+  const before = structuredClone(params);
+  const result = projectTurnSelectionRejection(params);
+  assert.equal(result.error_code,"turn_todo_selection_deferred");
+  assert.deepEqual(result.selection_rejection, {schema_version:"loopx_turn_selection_rejection_v0",
+    source:"quota.should-run",requested_todo_id:"todo_worker",state:"deferred",reason_code:"control_repair",
+    delivery_preemptions:["control_repair","delivery_not_allowed"],recovery_action:"reenter_guard_without_selection",
+    status_health_ok:false,contract_error_count:3});
+  assert.deepEqual(params,before);
+});
+
+test("absent or different selection stays unavailable, never borrowed or launchable", () => {
+  for (const raw of [undefined,{}, {...qualification,requested_todo_id:"other"},
+    {...qualification,state:"qualified"}, {...qualification,state:["deferred"]},
+    {...qualification,state:["rejected"]}, {...qualification,state:["unavailable"]},
+    {...qualification,state:["control_repair"]}]) {
+    const result = projectTurnSelectionRejection({...params,decision:{action_selection_qualification:raw}});
+    assert.equal(result.error_code,"turn_todo_selection_unavailable");
+    assert.equal((result.selection_rejection as Record<string,unknown>).reason_code,null);
+  }
+});
+
+test("only a decoded string state is a refusal; array states fail closed at the reader", () => {
+  for (const raw of ["deferred","rejected","unavailable"]) assert.equal(turnSelectionRejectionState(raw),raw);
+  for (const raw of [["deferred"],["rejected"],["unavailable"],[],["control_repair"],null,7,{state:"deferred"}])
+    assert.equal(turnSelectionRejectionState(raw),null);
+  const refusal = (state: unknown) => ({schema_version:"loopx_turn_selection_rejection_v0",source:"quota.should-run",
+    requested_todo_id:"todo_worker",state,reason_code:"control_repair",delivery_preemptions:["control_repair"],
+    recovery_action:"reenter_guard_without_selection",status_health_ok:false,contract_error_count:3});
+  for (const raw of ["deferred","rejected","unavailable"]) {
+    const read = readTurnSelectionRejection(refusal(raw),"todo_worker");
+    assert.equal(read.state,raw); assert.equal(Array.isArray(read.state),false);
+  }
+  for (const raw of [["deferred"],["rejected"],["unavailable"],[],null,7])
+    assert.throws(() => readTurnSelectionRejection(refusal(raw),"todo_worker"),/matching bounded/);
+});
+
+test("raw messages, paths and unsupported counts are not disclosed by the bounded projection", () => {
+  const result = projectTurnSelectionRejection({...params,contract_error_count:-1,decision:{...params.decision,
+    private_context:"not exported",action_selection_qualification:{...qualification,reason:"raw message",
+      delivery_preemptions:["control_repair","/operator/path"],recovery_action:"raw recovery"}}});
+  const refusal = result.selection_rejection as Record<string,unknown>;
+  assert.equal(refusal.reason_code,null); assert.equal(refusal.recovery_action,null);
+  assert.equal(refusal.contract_error_count,null); assert.deepEqual(refusal.delivery_preemptions,["control_repair"]);
+  assert.equal(JSON.stringify(result).includes("not exported"),false);
+});
diff --git a/tests/test_delegation_preflight.py b/tests/test_delegation_preflight.py
index 389f2a1645..433f284392 100644
--- a/tests/test_delegation_preflight.py
+++ b/tests/test_delegation_preflight.py
@@ -1,12 +1,15 @@
 """A binding inspection must use the actual Turn without launching or spending."""
 
 import json
+import asyncio
 import os
 import subprocess
 import sys
 from pathlib import Path
 
 import pytest
+from mcp import ClientSession, StdioServerParameters
+from mcp.client.stdio import stdio_client
 
 from loopx.control_plane.turn_driver import build_loopx_turn_plan
 from loopx.control_plane.turn_driver.executor import (
@@ -19,6 +22,59 @@
 service = delegation_service
 
 
+@pytest.mark.parametrize("health_repair", [False, True])
+def test_actual_workspace_scan_refusal_is_typed_and_effect_free(service, health_repair):
+    root, runner = service
+    if health_repair:
+        registry = json.loads(runner.registry.read_text())
+        registry["goals"][0]["control_plane"] = {"self_repair": {
+            "enabled": True, "allow_health_blocker_repair": True,
+        }}
+        runner.registry.write_text(json.dumps(registry))
+    workspace = Path(json.loads(runner.config.read_text())["bindings"][0]["workspace"])
+    # A synthetic literal exercises the real scanner; never exempt test files.
+    (workspace / "unsafe-fixture.py").write_text("fixture = " + repr("tok" + "en=" + "abcdefghijklmnop1234"))
+    before = runner.registry.read_bytes()
+    status, result = cli(runner, "inspect", "--binding-id", "analysis")
+    assert status == 0, result
+    assert result["state"] == "turn_blocked"
+    assert result["authority_ready"] and result["acceptance_ready"]
+    assert result["turn_eligible"] is False and result["executor"] is None
+    refusal = result["turn_blocker"]
+    assert refusal["requested_todo_id"] == "todo_analyst-initial"
+    assert refusal["state"] == "deferred"
+    assert refusal["reason_code"] == ("control_repair" if health_repair else "delivery_not_allowed")
+    assert refusal["status_health_ok"] is False and refusal["contract_error_count"] >= 1
+    assert not any(result["effects"].values())
+    if health_repair:
+        async def inspect_mcp():
+            params = StdioServerParameters(command=sys.executable, args=[
+                "-m", "loopx.collaboration_mcp", "--registry", str(runner.registry),
+                "--runtime-root", str(runner.root), "--goal-id", runner.goal_id,
+                "--agent-id", runner.agent_id, "--workspace", str(root / "lead"),
+                "--execution-config", str(runner.config),
+            ])
+            async with stdio_client(params) as (reader, writer):
+                async with ClientSession(reader, writer) as session:
+                    await session.initialize()
+                    inspected = await session.call_tool("inspect_execution_binding", {"binding_id": "analysis"})
+                    assert not inspected.isError
+                    return json.loads(inspected.content[0].text)
+        mcp_result = asyncio.run(inspect_mcp())
+        assert mcp_result["turn_blocker"] == refusal
+        assert mcp_result["state"] == "turn_blocked"
+        assert not any(mcp_result["effects"].values())
+    assert runner.registry.read_bytes() == before
+    assert not (root / "host-started").exists()
+    assert not list((root / "runtime" / "goals").glob("*/turns/*.json"))
+    # Removing this exact synthetic input restores ordinary inspection, rather
+    # than leaving a persisted repair route or choosing a different Todo.
+    (workspace / "unsafe-fixture.py").unlink()
+    status, restored = cli(runner, "inspect", "--binding-id", "analysis")
+    assert status == 0 and restored["turn_eligible"], restored
+    assert not any(restored["effects"].values())
+
+
 def test_real_cli_preflight_preserves_unknown_runtime_and_state(service):
     root, runner = service
     before = runner.registry.read_bytes()
@@ -116,6 +172,46 @@ def test_preflight_surfaces_actual_turn_rejection_without_launch(service):
     assert not (root / "host-started").exists()
 
 
+def test_real_runtime_decodes_only_string_selection_states():
+    """The live managed runtime rejects a JSON-array refusal enum instead of echoing it."""
+    from loopx.control_plane.effect_runtime import (
+        EffectRuntimeRemoteError,
+        effect_runtime_result,
+    )
+
+    binding = {"id": "review", "agent_id": "reviewer", "todo_id": "todo_review"}
+    effects = {"host_invoked": False, "state_written": False,
+               "quota_spent": False, "scheduler_acknowledged": False}
+
+    def params(state):
+        return {
+            "binding": binding,
+            "authority": {"ready": True, "reason": None},
+            "preview": {
+                "ok": False, "effects_scope": "current_invocation", "effects": effects,
+                "error_code": "turn_todo_selection_deferred",
+                "selection_rejection": {
+                    "schema_version": "loopx_turn_selection_rejection_v0",
+                    "source": "quota.should-run", "requested_todo_id": binding["todo_id"],
+                    "state": state, "reason_code": "control_repair",
+                    "delivery_preemptions": ["control_repair"],
+                    "recovery_action": "reenter_guard_without_selection",
+                    "status_health_ok": False, "contract_error_count": 2,
+                },
+            },
+            "acceptance": None, "validation_files_current": False,
+        }
+
+    accepted = effect_runtime_result("collaboration.delegation.preflight", params("deferred"))
+    assert accepted["state"] == "turn_blocked"
+    assert accepted["turn_blocker"]["state"] == "deferred"
+    assert accepted["turn_eligible"] is False and accepted["executor"] is None
+    assert not any(accepted["effects"].values())
+    for raw in (["deferred"], ["rejected"], ["unavailable"]):
+        with pytest.raises(EffectRuntimeRemoteError):
+            effect_runtime_result("collaboration.delegation.preflight", params(raw))
+
+
 def test_preflight_projects_unavailable_authority_without_turn_or_provider(
     service, monkeypatch
 ):
diff --git a/tests/test_loopx_turn_error_readback.py b/tests/test_loopx_turn_error_readback.py
index c5ebb3592a..db01c19f65 100644
--- a/tests/test_loopx_turn_error_readback.py
+++ b/tests/test_loopx_turn_error_readback.py
@@ -1,12 +1,46 @@
 """Error readback preserves uncertainty without inventing execution facts."""
 from __future__ import annotations
 
+import pytest
+
+from loopx.cli_commands.turn_decision import FreshTurnDecisionOwner, TurnTodoSelectionError
 from loopx.cli_commands.turn_rendering import (
     build_turn_error_payload,
     render_loopx_turn_execution_markdown,
+    render_loopx_turn_plan_markdown,
 )
 
 
+@pytest.mark.parametrize("command", ["plan", "run-once"])
+def test_selection_refusal_keeps_quota_reason_in_json_and_markdown(command):
+    calls = []
+
+    def current_decision(**kwargs):
+        calls.append(kwargs)
+        return {"status_health_ok": False, "action_selection_qualification": {
+            "requested_todo_id": "todo_fixture", "state": "deferred",
+            "reason": "control_repair", "recovery_action": "reenter_guard_without_selection",
+        }}
+
+    owner = FreshTurnDecisionOwner(
+        status_payload={"contract_summary": {"errors": 3}},
+        scheduler_execution_context={}, operator_inbox_urgency_projector=lambda **_: {},
+        build_turn_decision=current_decision, requested_todo_id="todo_fixture",
+    )
+    with pytest.raises(TurnTodoSelectionError) as caught:
+        owner.resolve()
+    assert calls == [{"requested_action_todo_id": "todo_fixture"}]
+    result = build_turn_error_payload({}, caught.value, turn_command=command)
+    assert result["error_code"] == "turn_todo_selection_deferred"
+    assert result["selection_rejection"]["requested_todo_id"] == "todo_fixture"
+    assert result["selection_rejection"]["reason_code"] == "control_repair"
+    assert result["selection_rejection"]["contract_error_count"] == 3
+    assert not any(result["effects"].values())
+    assert "journal_observation" not in result
+    renderer = render_loopx_turn_plan_markdown if command == "plan" else render_loopx_turn_execution_markdown
+    assert "reason=control_repair" in renderer(result)
+
+
 def test_pre_execution_error_preserves_hook_effects_without_claiming_a_host() -> None:
     result = build_turn_error_payload(
         {"effects": {"state_written": True}}, ValueError("invalid adapter"),