diff --git a/generate/candidate_operator.py b/generate/candidate_operator.py
index 8cd42dc3..2218b1bb 100644
--- a/generate/candidate_operator.py
+++ b/generate/candidate_operator.py
@@ -18,7 +18,7 @@ from generate.kernel_facts import SourceSpan
 from generate.search_gate import SearchGateDecision, SearchGateStatus
 
 CANDIDATE_OPERATOR_POLICY_VERSION = "candidate_operator.v1"
-CANDIDATE_OPERATOR_SET_VERSION = "candidate_operators.v1"
+CANDIDATE_OPERATOR_SET_VERSION = "candidate_operators.v2"
 
 MISSING_ROLE_CANDIDATE_OPERATOR_NAME = "missing_role_candidate"
 MISSING_ROLE_CANDIDATE_OPERATOR_VERSION = "missing_role_candidate.v1"
@@ -27,6 +27,12 @@ MISSING_ROLE_RESIDUAL_KIND = "missing_role"
 MISSING_ROLE_RESIDUAL_CODE = "direction_unbound"
 MISSING_ROLE_CANDIDATE_ORGAN = "unary_delta_transition"
 
+QUANTITY_ENTITY_BINDING_CANDIDATE_OPERATOR_NAME = "quantity_entity_binding_candidate"
+QUANTITY_ENTITY_BINDING_CANDIDATE_OPERATOR_VERSION = "quantity_entity_binding_candidate.v1"
+QUANTITY_ENTITY_BINDING_RESIDUAL_KIND = "missing_relation"
+QUANTITY_ENTITY_BINDING_RESIDUAL_CODE = "local_binding_relation_unbound"
+QUANTITY_ENTITY_BINDING_CANDIDATE_ORGAN = "quantity_entity_binding"
+
 _OPERATOR_FAMILY = "residual_missing_role"
 _SUPPORTED_DIRECTIONS = frozenset({"increase", "decrease"})
 _FIXED_BUDGET_CHARGE = BudgetCharge(candidates=1, steps=1)
@@ -75,6 +81,13 @@ class CandidateOperatorRefusalReason(str, Enum):
     MALFORMED_EVIDENCE_SPAN = "malformed_evidence_span"
     INVALID_OPERATOR_INPUT = "invalid_operator_input"
     UNSUPPORTED_SCHEMA_VERSION = "unsupported_schema_version"
+    INVALID_QUANTITY_ENTITY_CUE_TYPE = "invalid_quantity_entity_cue_type"
+    QUANTITY_ENTITY_CUE_COUNT_MISMATCH = "quantity_entity_cue_count_mismatch"
+    EMPTY_QUANTITY_MENTION_ID = "empty_quantity_mention_id"
+    EMPTY_ENTITY_MENTION_ID = "empty_entity_mention_id"
+    EMPTY_QUANTITY_KIND = "empty_quantity_kind"
+    EMPTY_UNIT_MENTION_ID = "empty_unit_mention_id"
+    MISSING_EVIDENCE_SPANS = "missing_evidence_spans"
 
 
 @dataclass(frozen=True, slots=True)
@@ -111,12 +124,38 @@ MISSING_ROLE_OPERATOR_POLICY = CandidateOperatorPolicy(
 )
 
 
+QUANTITY_ENTITY_BINDING_OPERATOR_POLICY = CandidateOperatorPolicy(
+    operator_policy_version=CANDIDATE_OPERATOR_POLICY_VERSION,
+    operator_family="residual_missing_relation",
+    operator_name=QUANTITY_ENTITY_BINDING_CANDIDATE_OPERATOR_NAME,
+    operator_version=QUANTITY_ENTITY_BINDING_CANDIDATE_OPERATOR_VERSION,
+    allowed_residual_kinds=(QUANTITY_ENTITY_BINDING_RESIDUAL_KIND,),
+    allowed_residual_codes=(QUANTITY_ENTITY_BINDING_RESIDUAL_CODE,),
+    allowed_candidate_organs=(QUANTITY_ENTITY_BINDING_CANDIDATE_ORGAN,),
+    max_attempts_per_run=1,
+    budget_charge=_FIXED_BUDGET_CHARGE,
+    depth=_OPERATOR_DEPTH,
+    max_parallelism=_OPERATOR_MAX_PARALLELISM,
+    determinism_requirements=_DETERMINISM_REQUIREMENTS,
+    forbidden_authority_paths=_FORBIDDEN_AUTHORITY_PATHS,
+)
+
+
 @dataclass(frozen=True, slots=True)
 class GroundedUnaryDeltaCue:
     direction: str
     evidence_spans: tuple[SourceSpan, ...]
 
 
+@dataclass(frozen=True, slots=True)
+class GroundedQuantityEntityCue:
+    quantity_mention_id: str
+    entity_mention_id: str
+    quantity_kind: str
+    evidence_spans: tuple[SourceSpan, ...]
+    unit_mention_id: str | None = None
+
+
 @dataclass(frozen=True, slots=True)
 class CandidateOperatorInput:
     input_digest: str
@@ -300,7 +339,28 @@ def _operator_set_table_payload() -> dict[str, object]:
                 },
                 "depth": MISSING_ROLE_OPERATOR_POLICY.depth,
                 "max_parallelism": MISSING_ROLE_OPERATOR_POLICY.max_parallelism,
-            }
+            },
+            {
+                "operator_family": QUANTITY_ENTITY_BINDING_OPERATOR_POLICY.operator_family,
+                "operator_name": QUANTITY_ENTITY_BINDING_OPERATOR_POLICY.operator_name,
+                "operator_version": QUANTITY_ENTITY_BINDING_OPERATOR_POLICY.operator_version,
+                "allowed_residual_kinds": list(
+                    QUANTITY_ENTITY_BINDING_OPERATOR_POLICY.allowed_residual_kinds
+                ),
+                "allowed_residual_codes": list(
+                    QUANTITY_ENTITY_BINDING_OPERATOR_POLICY.allowed_residual_codes
+                ),
+                "allowed_candidate_organs": list(
+                    QUANTITY_ENTITY_BINDING_OPERATOR_POLICY.allowed_candidate_organs
+                ),
+                "max_attempts_per_run": QUANTITY_ENTITY_BINDING_OPERATOR_POLICY.max_attempts_per_run,
+                "budget_charge": {
+                    "candidates": QUANTITY_ENTITY_BINDING_OPERATOR_POLICY.budget_charge.candidates,
+                    "steps": QUANTITY_ENTITY_BINDING_OPERATOR_POLICY.budget_charge.steps,
+                },
+                "depth": QUANTITY_ENTITY_BINDING_OPERATOR_POLICY.depth,
+                "max_parallelism": QUANTITY_ENTITY_BINDING_OPERATOR_POLICY.max_parallelism,
+            },
         ],
         "schema_versions": [],
         "policy_versions": [],
@@ -323,6 +383,22 @@ def _candidate_payload(direction: str) -> tuple[tuple[str, str], ...]:
     )
 
 
+def _quantity_entity_binding_candidate_payload(
+    cue: GroundedQuantityEntityCue,
+) -> tuple[tuple[str, str], ...]:
+    return (
+        ("binding_type", "quantity_entity"),
+        ("candidate_organ", QUANTITY_ENTITY_BINDING_CANDIDATE_ORGAN),
+        ("entity_mention_id", cue.entity_mention_id),
+        ("kind", "mention_binding"),
+        ("quantity_kind", cue.quantity_kind),
+        ("quantity_mention_id", cue.quantity_mention_id),
+        ("relation_type", "quantity_entity"),
+        ("source", "GroundedQuantityEntityCue"),
+        ("unit_mention_id", cue.unit_mention_id or ""),
+    )
+
+
 def _candidate_payload_dict(
     payload: tuple[tuple[str, str], ...],
 ) -> list[list[str]]:
@@ -925,6 +1001,400 @@ def build_missing_role_candidate(
     )
 
 
+def build_quantity_entity_binding_candidate(
+    *,
+    residual: object,
+    search_gate: SearchGateDecision,
+    compute_budget: ComputeBudgetDecision,
+    run: GeometricSearchRun,
+    problem_frame_digest: str,
+    original_contract_assessment_id: str,
+    grounded_quantity_entity_cues: tuple[GroundedQuantityEntityCue, ...],
+    attempt_index: int = 0,
+    schema_versions: tuple[tuple[str, str], ...] = (),
+    policy_versions: tuple[tuple[str, str], ...] = (),
+    explanation: str = "",
+) -> CandidateOperatorOutcome:
+    """Construct one quantity-entity binding candidate or return a typed refusal."""
+
+    operator_set_id = candidate_operator_set_id()
+    reasons: list[str] = []
+
+    residual_id = _safe_getattr(residual, "residual_id")
+    residual_kind = _enum_value(_safe_getattr(residual, "residual_kind"))
+    residual_code = _safe_getattr(residual, "residual_code")
+    candidate_organ = _safe_getattr(residual, "candidate_organ")
+    residual_spans = _safe_getattr(residual, "evidence_spans")
+
+    gate_id = _safe_getattr(search_gate, "decision_id")
+    gate_status = _safe_getattr(search_gate, "status")
+    gate_reason = _safe_getattr(search_gate, "reason_code")
+    gate_residual_ids = _safe_getattr(search_gate, "residual_ids")
+    gate_candidate_organ = _safe_getattr(search_gate, "candidate_organ")
+
+    budget_id = _safe_getattr(compute_budget, "budget_id")
+    budget_status = _safe_getattr(compute_budget, "status")
+    budget_reason = _safe_getattr(compute_budget, "reason_code")
+    budget_gate_id = _safe_getattr(compute_budget, "gate_decision_id")
+    max_candidates = _safe_getattr(compute_budget, "max_candidates")
+    max_depth = _safe_getattr(compute_budget, "max_depth")
+    max_steps = _safe_getattr(compute_budget, "max_steps")
+    max_parallelism = _safe_getattr(compute_budget, "max_parallelism")
+
+    run_id = _safe_getattr(run, "run_id")
+    run_gate_id = _safe_getattr(run, "gate_decision_id")
+    run_budget_id = _safe_getattr(run, "budget_id")
+    run_operator_set_id = _safe_getattr(run, "operator_set_id")
+    run_operator_set_version = _safe_getattr(run, "operator_set_version")
+    run_problem_frame_digest = _safe_getattr(run, "problem_frame_digest")
+    run_assessment_id = _safe_getattr(run, "contract_assessment_id")
+    run_residual_ids = _safe_getattr(run, "residual_ids")
+
+    if not _valid_residual_record(residual):
+        reasons.append(CandidateOperatorRefusalReason.INVALID_OPERATOR_INPUT.value)
+    if not isinstance(search_gate, SearchGateDecision):
+        reasons.append(CandidateOperatorRefusalReason.INVALID_OPERATOR_INPUT.value)
+    if not isinstance(compute_budget, ComputeBudgetDecision):
+        reasons.append(CandidateOperatorRefusalReason.INVALID_OPERATOR_INPUT.value)
+    if not isinstance(run, GeometricSearchRun):
+        reasons.append(CandidateOperatorRefusalReason.INVALID_OPERATOR_INPUT.value)
+
+    if not _nonempty_text(residual_id):
+        reasons.append(CandidateOperatorRefusalReason.INVALID_OPERATOR_INPUT.value)
+    if residual_kind != QUANTITY_ENTITY_BINDING_RESIDUAL_KIND:
+        reasons.append(CandidateOperatorRefusalReason.UNSUPPORTED_RESIDUAL_KIND.value)
+    if residual_code != QUANTITY_ENTITY_BINDING_RESIDUAL_CODE:
+        reasons.append(CandidateOperatorRefusalReason.UNSUPPORTED_RESIDUAL_CODE.value)
+    if candidate_organ != QUANTITY_ENTITY_BINDING_CANDIDATE_ORGAN:
+        reasons.append(CandidateOperatorRefusalReason.UNSUPPORTED_CANDIDATE_ORGAN.value)
+    if not _valid_spans(residual_spans):
+        reasons.append(CandidateOperatorRefusalReason.MALFORMED_EVIDENCE_SPAN.value)
+
+    if gate_status is not SearchGateStatus.ELIGIBLE:
+        reasons.append(CandidateOperatorRefusalReason.INELIGIBLE_SEARCH_GATE.value)
+    if gate_reason != "eligible_missing_relation":
+        reasons.append(CandidateOperatorRefusalReason.INELIGIBLE_SEARCH_GATE.value)
+    if gate_candidate_organ != QUANTITY_ENTITY_BINDING_CANDIDATE_ORGAN:
+        reasons.append(CandidateOperatorRefusalReason.UNSUPPORTED_CANDIDATE_ORGAN.value)
+
+    if budget_status is not ComputeBudgetStatus.BUDGET_ALLOWED:
+        reasons.append(CandidateOperatorRefusalReason.NON_ALLOWED_BUDGET.value)
+    if budget_reason != "budget_allowed_missing_relation":
+        reasons.append(CandidateOperatorRefusalReason.NON_ALLOWED_BUDGET.value)
+
+    if not _nonempty_text(budget_gate_id) or not _nonempty_text(gate_id):
+        reasons.append(CandidateOperatorRefusalReason.GATE_BUDGET_MISMATCH.value)
+    elif budget_gate_id != gate_id:
+        reasons.append(CandidateOperatorRefusalReason.GATE_BUDGET_MISMATCH.value)
+
+    if not _nonempty_text(run_gate_id) or not _nonempty_text(gate_id):
+        reasons.append(CandidateOperatorRefusalReason.RUN_GATE_MISMATCH.value)
+    elif run_gate_id != gate_id:
+        reasons.append(CandidateOperatorRefusalReason.RUN_GATE_MISMATCH.value)
+
+    if not _nonempty_text(run_budget_id) or not _nonempty_text(budget_id):
+        reasons.append(CandidateOperatorRefusalReason.RUN_BUDGET_MISMATCH.value)
+    elif run_budget_id != budget_id:
+        reasons.append(CandidateOperatorRefusalReason.RUN_BUDGET_MISMATCH.value)
+
+    if run_operator_set_id != operator_set_id:
+        reasons.append(CandidateOperatorRefusalReason.OPERATOR_SET_MISMATCH.value)
+    if run_operator_set_version != CANDIDATE_OPERATOR_SET_VERSION:
+        reasons.append(CandidateOperatorRefusalReason.OPERATOR_SET_MISMATCH.value)
+
+    if type(attempt_index) is not int or attempt_index < 0:
+        reasons.append(CandidateOperatorRefusalReason.INVALID_OPERATOR_INPUT.value)
+    elif attempt_index != 0:
+        reasons.append(CandidateOperatorRefusalReason.ATTEMPT_INDEX_EXCEEDS_BUDGET.value)
+        
+    if type(max_candidates) is not int:
+        reasons.append(CandidateOperatorRefusalReason.INVALID_OPERATOR_INPUT.value)
+    elif max_candidates < 1:
+        reasons.append(CandidateOperatorRefusalReason.BUDGET_CHARGE_EXCEEDS_BUDGET.value)
+
+    if attempt_index >= QUANTITY_ENTITY_BINDING_OPERATOR_POLICY.max_attempts_per_run:
+        reasons.append(
+            CandidateOperatorRefusalReason.ATTEMPT_INDEX_EXCEEDS_OPERATOR_POLICY.value
+        )
+
+    if type(max_depth) is not int or type(max_steps) is not int:
+        reasons.append(CandidateOperatorRefusalReason.INVALID_OPERATOR_INPUT.value)
+    else:
+        charge = QUANTITY_ENTITY_BINDING_OPERATOR_POLICY.budget_charge
+        if charge.candidates > max_candidates:
+            reasons.append(
+                CandidateOperatorRefusalReason.BUDGET_CHARGE_EXCEEDS_BUDGET.value
+            )
+        if charge.steps > max_steps:
+            reasons.append(
+                CandidateOperatorRefusalReason.BUDGET_CHARGE_EXCEEDS_BUDGET.value
+            )
+        if QUANTITY_ENTITY_BINDING_OPERATOR_POLICY.depth > max_depth:
+            reasons.append(
+                CandidateOperatorRefusalReason.BUDGET_CHARGE_EXCEEDS_BUDGET.value
+            )
+
+    if type(max_parallelism) is not int or max_parallelism != 1:
+        reasons.append(CandidateOperatorRefusalReason.NON_SERIAL_BUDGET.value)
+
+    if not _valid_version_pairs(schema_versions):
+        reasons.append(CandidateOperatorRefusalReason.UNSUPPORTED_SCHEMA_VERSION.value)
+    if not _valid_version_pairs(policy_versions):
+        reasons.append(CandidateOperatorRefusalReason.UNSUPPORTED_SCHEMA_VERSION.value)
+
+    if not _sha256_text(problem_frame_digest):
+        reasons.append(CandidateOperatorRefusalReason.INVALID_OPERATOR_INPUT.value)
+    if not _nonempty_text(original_contract_assessment_id):
+        reasons.append(CandidateOperatorRefusalReason.INVALID_OPERATOR_INPUT.value)
+    if not _sha256_text(run_id):
+        reasons.append(CandidateOperatorRefusalReason.INVALID_OPERATOR_INPUT.value)
+
+    if (
+        isinstance(residual_id, str)
+        and isinstance(gate_residual_ids, tuple)
+        and residual_id not in gate_residual_ids
+    ):
+        reasons.append(CandidateOperatorRefusalReason.INVALID_OPERATOR_INPUT.value)
+    if (
+        isinstance(residual_id, str)
+        and isinstance(run_residual_ids, tuple)
+        and residual_id not in run_residual_ids
+    ):
+        reasons.append(CandidateOperatorRefusalReason.INVALID_OPERATOR_INPUT.value)
+    if (
+        isinstance(problem_frame_digest, str)
+        and isinstance(run_problem_frame_digest, str)
+        and problem_frame_digest != run_problem_frame_digest
+    ):
+        reasons.append(CandidateOperatorRefusalReason.INVALID_OPERATOR_INPUT.value)
+    if (
+        isinstance(original_contract_assessment_id, str)
+        and isinstance(run_assessment_id, str)
+        and original_contract_assessment_id != run_assessment_id
+    ):
+        reasons.append(CandidateOperatorRefusalReason.INVALID_OPERATOR_INPUT.value)
+
+    if not isinstance(grounded_quantity_entity_cues, tuple):
+        reasons.append(CandidateOperatorRefusalReason.INVALID_QUANTITY_ENTITY_CUE_TYPE.value)
+    elif len(grounded_quantity_entity_cues) == 0:
+        reasons.append(CandidateOperatorRefusalReason.MISSING_TYPED_CUE.value)
+    elif len(grounded_quantity_entity_cues) > 1:
+        reasons.append(CandidateOperatorRefusalReason.QUANTITY_ENTITY_CUE_COUNT_MISMATCH.value)
+
+    cue: GroundedQuantityEntityCue | None = None
+    if isinstance(grounded_quantity_entity_cues, tuple) and len(grounded_quantity_entity_cues) == 1:
+        cue = grounded_quantity_entity_cues[0]
+        if not isinstance(cue, GroundedQuantityEntityCue):
+            reasons.append(CandidateOperatorRefusalReason.INVALID_QUANTITY_ENTITY_CUE_TYPE.value)
+        else:
+            if not _nonempty_text(cue.quantity_mention_id):
+                reasons.append(CandidateOperatorRefusalReason.EMPTY_QUANTITY_MENTION_ID.value)
+            if not _nonempty_text(cue.entity_mention_id):
+                reasons.append(CandidateOperatorRefusalReason.EMPTY_ENTITY_MENTION_ID.value)
+            if not _nonempty_text(cue.quantity_kind):
+                reasons.append(CandidateOperatorRefusalReason.EMPTY_QUANTITY_KIND.value)
+            if cue.unit_mention_id is not None and not _nonempty_text(cue.unit_mention_id):
+                reasons.append(CandidateOperatorRefusalReason.EMPTY_UNIT_MENTION_ID.value)
+
+            if not _valid_spans(cue.evidence_spans):
+                reasons.append(CandidateOperatorRefusalReason.MALFORMED_EVIDENCE_SPAN.value)
+            elif not cue.evidence_spans:
+                reasons.append(CandidateOperatorRefusalReason.MISSING_EVIDENCE_SPANS.value)
+            elif isinstance(residual_spans, tuple) and not _spans_grounded_in_residual(
+                cue.evidence_spans, residual_spans
+            ):
+                reasons.append(CandidateOperatorRefusalReason.MALFORMED_EVIDENCE_SPAN.value)
+
+    input_digest: str | None = None
+    if (
+        not reasons
+        and isinstance(residual_id, str)
+        and isinstance(run_id, str)
+        and isinstance(gate_id, str)
+        and isinstance(budget_id, str)
+        and type(attempt_index) is int
+    ):
+        input_digest = _canonical_digest(
+            _input_digest_payload(
+                operator_policy_version=CANDIDATE_OPERATOR_POLICY_VERSION,
+                operator_name=QUANTITY_ENTITY_BINDING_CANDIDATE_OPERATOR_NAME,
+                operator_version=QUANTITY_ENTITY_BINDING_CANDIDATE_OPERATOR_VERSION,
+                problem_frame_digest=problem_frame_digest,
+                original_contract_assessment_id=original_contract_assessment_id,
+                residual_id=residual_id,
+                residual_kind=QUANTITY_ENTITY_BINDING_RESIDUAL_KIND,
+                residual_code=QUANTITY_ENTITY_BINDING_RESIDUAL_CODE,
+                candidate_organ=QUANTITY_ENTITY_BINDING_CANDIDATE_ORGAN,
+                search_gate_decision_id=gate_id,
+                compute_budget_id=budget_id,
+                geometric_search_run_id=run_id,
+                operator_set_id=operator_set_id,
+                operator_set_version=CANDIDATE_OPERATOR_SET_VERSION,
+                attempt_index=attempt_index,
+                schema_versions=schema_versions,
+                policy_versions=policy_versions,
+            )
+        )
+
+    if reasons:
+        operator_refusal_id = _canonical_digest(
+            _operator_refusal_id_payload(
+                operator_policy_version=CANDIDATE_OPERATOR_POLICY_VERSION,
+                input_digest=input_digest,
+                geometric_search_run_id=_text_or_none(run_id),
+                residual_id=_text_or_none(residual_id),
+                operator_name=QUANTITY_ENTITY_BINDING_CANDIDATE_OPERATOR_NAME,
+                reason_codes=tuple(dict.fromkeys(reasons)),
+            )
+        )
+        return CandidateOperatorRefusal(
+            operator_refusal_id=operator_refusal_id,
+            operator_policy_version=CANDIDATE_OPERATOR_POLICY_VERSION,
+            input_digest=input_digest,
+            geometric_search_run_id=_text_or_none(run_id),
+            residual_id=_text_or_none(residual_id),
+            operator_name=QUANTITY_ENTITY_BINDING_CANDIDATE_OPERATOR_NAME,
+            reason_codes=tuple(dict.fromkeys(reasons)),
+            explanation="Candidate operator refused: " + ", ".join(dict.fromkeys(reasons)) + ".",
+        )
+
+    assert cue is not None
+    assert input_digest is not None
+    assert isinstance(residual_id, str)
+    assert isinstance(run_id, str)
+
+    evidence_spans = cue.evidence_spans
+    payload = _quantity_entity_binding_candidate_payload(cue)
+    candidate_digest = _canonical_digest(
+        _candidate_digest_payload(
+            problem_frame_digest=problem_frame_digest,
+            candidate_organ=QUANTITY_ENTITY_BINDING_CANDIDATE_ORGAN,
+            candidate_payload=payload,
+            evidence_spans=evidence_spans,
+        )
+    )
+
+    attempt_id = _canonical_digest(
+        _attempt_id_payload(
+            attempt_index=attempt_index,
+            parent_attempt_id=None,
+            operator_id=QUANTITY_ENTITY_BINDING_CANDIDATE_OPERATOR_NAME,
+            operator_version=QUANTITY_ENTITY_BINDING_CANDIDATE_OPERATOR_VERSION,
+            input_digest=input_digest,
+            candidate_digest=candidate_digest,
+            budget_charge=_FIXED_BUDGET_CHARGE,
+            depth=_OPERATOR_DEPTH,
+            step_index=attempt_index,
+            replay_status=CandidateReplayStatus.REPLAY_PENDING,
+            replay_blockers=(),
+            evidence_spans=evidence_spans,
+        )
+    )
+
+    candidate_reconstruction_digest = _canonical_digest(
+        _reconstruction_digest_payload(
+            geometric_search_run_id=run_id,
+            attempt_id=attempt_id,
+            attempt_index=attempt_index,
+            operator_set_id=operator_set_id,
+            operator_set_version=CANDIDATE_OPERATOR_SET_VERSION,
+            operator_name=QUANTITY_ENTITY_BINDING_CANDIDATE_OPERATOR_NAME,
+            operator_version=QUANTITY_ENTITY_BINDING_CANDIDATE_OPERATOR_VERSION,
+            residual_id=residual_id,
+            residual_kind=QUANTITY_ENTITY_BINDING_RESIDUAL_KIND,
+            residual_code=QUANTITY_ENTITY_BINDING_RESIDUAL_CODE,
+            candidate_digest=candidate_digest,
+            evidence_spans=evidence_spans,
+            schema_versions=schema_versions,
+            policy_versions=policy_versions,
+        )
+    )
+
+    operator_provenance = (
+        ("operator_family", QUANTITY_ENTITY_BINDING_OPERATOR_POLICY.operator_family),
+        ("operator_name", QUANTITY_ENTITY_BINDING_CANDIDATE_OPERATOR_NAME),
+        ("operator_version", QUANTITY_ENTITY_BINDING_CANDIDATE_OPERATOR_VERSION),
+        ("operator_set_id", operator_set_id),
+        ("operator_set_version", CANDIDATE_OPERATOR_SET_VERSION),
+    )
+
+    candidate_reconstruction = CandidateReconstruction(
+        candidate_digest=candidate_digest,
+        candidate_reconstruction_digest=candidate_reconstruction_digest,
+        candidate_organ=QUANTITY_ENTITY_BINDING_CANDIDATE_ORGAN,
+        candidate_payload=payload,
+        evidence_spans=evidence_spans,
+        operator_name=QUANTITY_ENTITY_BINDING_CANDIDATE_OPERATOR_NAME,
+        operator_version=QUANTITY_ENTITY_BINDING_CANDIDATE_OPERATOR_VERSION,
+        operator_provenance=operator_provenance,
+        source_residual_id=residual_id,
+        problem_frame_digest=problem_frame_digest,
+        original_contract_assessment_id=original_contract_assessment_id,
+    )
+
+    candidate_attempt = CandidateAttempt(
+        attempt_id=attempt_id,
+        attempt_index=attempt_index,
+        parent_attempt_id=None,
+        operator_id=QUANTITY_ENTITY_BINDING_CANDIDATE_OPERATOR_NAME,
+        operator_version=QUANTITY_ENTITY_BINDING_CANDIDATE_OPERATOR_VERSION,
+        input_digest=input_digest,
+        candidate_digest=candidate_digest,
+        budget_charge=_FIXED_BUDGET_CHARGE,
+        depth=_OPERATOR_DEPTH,
+        step_index=attempt_index,
+        replay_status=CandidateReplayStatus.REPLAY_PENDING,
+        replay_blockers=(),
+        evidence_spans=evidence_spans,
+        explanation="",
+    )
+
+    reason_codes: tuple[str, ...] = ()
+    operator_result_id = _canonical_digest(
+        _operator_result_id_payload(
+            operator_policy_version=CANDIDATE_OPERATOR_POLICY_VERSION,
+            input_digest=input_digest,
+            geometric_search_run_id=run_id,
+            attempt_id=attempt_id,
+            attempt_index=attempt_index,
+            candidate_digest=candidate_digest,
+            candidate_reconstruction_digest=candidate_reconstruction_digest,
+            candidate_organ=QUANTITY_ENTITY_BINDING_CANDIDATE_ORGAN,
+            operator_name=QUANTITY_ENTITY_BINDING_CANDIDATE_OPERATOR_NAME,
+            operator_version=QUANTITY_ENTITY_BINDING_CANDIDATE_OPERATOR_VERSION,
+            reason_codes=reason_codes,
+            evidence_spans=evidence_spans,
+        )
+    )
+
+    safe_explanation = explanation or (
+        "Quantity-entity binding candidate constructed for residual "
+        + residual_id
+        + " linking "
+        + cue.quantity_mention_id
+        + " and "
+        + cue.entity_mention_id
+        + "."
+    )
+    return CandidateOperatorResult(
+        operator_result_id=operator_result_id,
+        operator_policy_version=CANDIDATE_OPERATOR_POLICY_VERSION,
+        input_digest=input_digest,
+        geometric_search_run_id=run_id,
+        attempt_id=attempt_id,
+        attempt_index=attempt_index,
+        candidate_digest=candidate_digest,
+        candidate_reconstruction_digest=candidate_reconstruction_digest,
+        candidate_organ=QUANTITY_ENTITY_BINDING_CANDIDATE_ORGAN,
+        operator_name=QUANTITY_ENTITY_BINDING_CANDIDATE_OPERATOR_NAME,
+        operator_version=QUANTITY_ENTITY_BINDING_CANDIDATE_OPERATOR_VERSION,
+        candidate_attempt=candidate_attempt,
+        candidate_reconstruction=candidate_reconstruction,
+        reason_codes=reason_codes,
+        evidence_spans=evidence_spans,
+        explanation=safe_explanation,
+    )
+
 __all__ = [
     "CANDIDATE_OPERATOR_POLICY_VERSION",
     "CANDIDATE_OPERATOR_SET_VERSION",
@@ -933,9 +1403,15 @@ __all__ = [
     "MISSING_ROLE_RESIDUAL_KIND",
     "MISSING_ROLE_RESIDUAL_CODE",
     "MISSING_ROLE_CANDIDATE_ORGAN",
+    "QUANTITY_ENTITY_BINDING_CANDIDATE_OPERATOR_NAME",
+    "QUANTITY_ENTITY_BINDING_CANDIDATE_OPERATOR_VERSION",
+    "QUANTITY_ENTITY_BINDING_RESIDUAL_KIND",
+    "QUANTITY_ENTITY_BINDING_RESIDUAL_CODE",
+    "QUANTITY_ENTITY_BINDING_CANDIDATE_ORGAN",
     "CandidateOperatorRefusalReason",
     "CandidateOperatorPolicy",
     "GroundedUnaryDeltaCue",
+    "GroundedQuantityEntityCue",
     "CandidateOperatorInput",
     "CandidateReconstruction",
     "CandidateOperatorResult",
@@ -943,4 +1419,5 @@ __all__ = [
     "CandidateOperatorOutcome",
     "candidate_operator_set_id",
     "build_missing_role_candidate",
-]
\ No newline at end of file
+    "build_quantity_entity_binding_candidate",
+]
diff --git a/tests/test_candidate_operator.py b/tests/test_candidate_operator.py
index 5f992438..1a9121d8 100644
--- a/tests/test_candidate_operator.py
+++ b/tests/test_candidate_operator.py
@@ -237,7 +237,19 @@ def _expected_operator_set_id() -> str:
                 "budget_charge": {"candidates": 1, "steps": 1},
                 "depth": 1,
                 "max_parallelism": 1,
-            }
+            },
+            {
+                "operator_family": "residual_missing_relation",
+                "operator_name": "quantity_entity_binding_candidate",
+                "operator_version": "quantity_entity_binding_candidate.v1",
+                "allowed_residual_kinds": ["missing_relation"],
+                "allowed_residual_codes": ["local_binding_relation_unbound"],
+                "allowed_candidate_organs": ["quantity_entity_binding"],
+                "max_attempts_per_run": 1,
+                "budget_charge": {"candidates": 1, "steps": 1},
+                "depth": 1,
+                "max_parallelism": 1,
+            },
         ],
         "schema_versions": [],
         "policy_versions": [],
@@ -412,7 +424,7 @@ def test_public_api_exports_are_exact() -> None:
     import generate.candidate_operator as candidate_operator
 
     assert CANDIDATE_OPERATOR_POLICY_VERSION == "candidate_operator.v1"
-    assert CANDIDATE_OPERATOR_SET_VERSION == "candidate_operators.v1"
+    assert CANDIDATE_OPERATOR_SET_VERSION == "candidate_operators.v2"
     assert MISSING_ROLE_CANDIDATE_OPERATOR_NAME == "missing_role_candidate"
     assert MISSING_ROLE_CANDIDATE_OPERATOR_VERSION == "missing_role_candidate.v1"
     assert MISSING_ROLE_RESIDUAL_KIND == "missing_role"
@@ -426,9 +438,15 @@ def test_public_api_exports_are_exact() -> None:
         "MISSING_ROLE_RESIDUAL_KIND",
         "MISSING_ROLE_RESIDUAL_CODE",
         "MISSING_ROLE_CANDIDATE_ORGAN",
+        "QUANTITY_ENTITY_BINDING_CANDIDATE_OPERATOR_NAME",
+        "QUANTITY_ENTITY_BINDING_CANDIDATE_OPERATOR_VERSION",
+        "QUANTITY_ENTITY_BINDING_RESIDUAL_KIND",
+        "QUANTITY_ENTITY_BINDING_RESIDUAL_CODE",
+        "QUANTITY_ENTITY_BINDING_CANDIDATE_ORGAN",
         "CandidateOperatorRefusalReason",
         "CandidateOperatorPolicy",
         "GroundedUnaryDeltaCue",
+        "GroundedQuantityEntityCue",
         "CandidateOperatorInput",
         "CandidateReconstruction",
         "CandidateOperatorResult",
@@ -436,6 +454,7 @@ def test_public_api_exports_are_exact() -> None:
         "CandidateOperatorOutcome",
         "candidate_operator_set_id",
         "build_missing_role_candidate",
+        "build_quantity_entity_binding_candidate",
     )
 
 
diff --git a/tests/test_candidate_operator_quantity_entity_binding.py b/tests/test_candidate_operator_quantity_entity_binding.py
new file mode 100644
index 00000000..47c9c3f2
--- /dev/null
+++ b/tests/test_candidate_operator_quantity_entity_binding.py
@@ -0,0 +1,551 @@
+from __future__ import annotations
+
+import hashlib
+import json
+from types import SimpleNamespace
+
+from generate.candidate_operator import (
+    CANDIDATE_OPERATOR_SET_VERSION,
+    QUANTITY_ENTITY_BINDING_CANDIDATE_OPERATOR_NAME,
+    QUANTITY_ENTITY_BINDING_CANDIDATE_OPERATOR_VERSION,
+    QUANTITY_ENTITY_BINDING_CANDIDATE_ORGAN,
+    QUANTITY_ENTITY_BINDING_RESIDUAL_CODE,
+    CandidateOperatorRefusal,
+    CandidateOperatorRefusalReason,
+    CandidateOperatorResult,
+    GroundedQuantityEntityCue,
+    build_quantity_entity_binding_candidate,
+    candidate_operator_set_id,
+)
+from generate.compute_budget import ComputeBudgetDecision, ComputeBudgetStatus
+from generate.contract_residuals import ContractResidual, ResidualKind, ResidualSourceAxis
+from generate.geometric_search_run import (
+    BudgetCharge,
+    CandidateReplayStatus,
+    GeometricSearchRun,
+    initialize_geometric_search_run,
+)
+from generate.kernel_facts import SourceSpan
+from generate.search_gate import SearchGateDecision, SearchGateStatus
+from generate.run_attempt_binding import bind_candidate_attempt_to_run
+from generate.replay_adapter import build_replay_adapter_input_from_binding, ReplayAdapterRefusal
+
+
+def _digest(payload: dict[str, object]) -> str:
+    encoded = json.dumps(
+        payload,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+    ).encode("utf-8")
+    return hashlib.sha256(encoded).hexdigest()
+
+
+def _span_payload(span: SourceSpan) -> dict[str, object]:
+    return {
+        "text": span.text,
+        "start": span.start,
+        "end": span.end,
+        "sentence_index": span.sentence_index,
+    }
+
+
+def _span(text: str = "5 apples", start: int = 10, end: int = 18) -> SourceSpan:
+    return SourceSpan(text=text, start=start, end=end, sentence_index=0)
+
+
+def _residual(
+    *,
+    residual_id: str | None = None,
+    evidence_spans: tuple[SourceSpan, ...] | None = None,
+    residual_kind: ResidualKind = ResidualKind.MISSING_RELATION,
+    residual_code: str = QUANTITY_ENTITY_BINDING_RESIDUAL_CODE,
+    candidate_organ: str = QUANTITY_ENTITY_BINDING_CANDIDATE_ORGAN,
+) -> ContractResidual:
+    spans = evidence_spans if evidence_spans is not None else (_span(),)
+    if residual_id is None:
+        payload = {
+            "candidate_organ": candidate_organ,
+            "residual_kind": residual_kind.value,
+            "residual_code": residual_code,
+            "evidence_spans": [_span_payload(span) for span in spans],
+        }
+        residual_id = _digest(payload)
+    return ContractResidual(
+        residual_id=residual_id,
+        candidate_organ=candidate_organ,
+        family_id="state_change.quantity_entity",
+        residual_kind=residual_kind,
+        residual_code=residual_code,
+        source_axis=ResidualSourceAxis.ROLE,
+        evidence_spans=spans,
+        explanation="residual prose",
+    )
+
+
+def _gate(
+    residual: ContractResidual,
+    *,
+    status: SearchGateStatus = SearchGateStatus.ELIGIBLE,
+    reason_code: str = "eligible_missing_relation",
+    input_digest: str = "a" * 64,
+) -> SearchGateDecision:
+    payload = {
+        "policy_version": "search_gate.v1",
+        "input_digest": input_digest,
+        "residual_ids": [residual.residual_id],
+        "candidate_organ": residual.candidate_organ,
+        "status": status.value,
+        "reason_code": reason_code,
+        "evidence_spans": [_span_payload(span) for span in residual.evidence_spans],
+    }
+    return SearchGateDecision(
+        decision_id=_digest(payload),
+        policy_version="search_gate.v1",
+        input_digest=input_digest,
+        residual_ids=(residual.residual_id,),
+        candidate_organ=residual.candidate_organ,
+        status=status,
+        reason_code=reason_code,
+        evidence_spans=residual.evidence_spans,
+        explanation="gate prose",
+    )
+
+
+def _budget(
+    gate: SearchGateDecision,
+    *,
+    status: ComputeBudgetStatus = ComputeBudgetStatus.BUDGET_ALLOWED,
+    reason_code: str = "budget_allowed_missing_relation",
+    max_candidates: int = 5,
+    max_depth: int = 2,
+    max_steps: int = 10,
+    max_parallelism: int = 1,
+) -> ComputeBudgetDecision:
+    payload = {
+        "policy_version": "compute_budget.v1",
+        "gate_decision_id": gate.decision_id,
+        "gate_policy_version": gate.policy_version,
+        "gate_input_digest": gate.input_digest,
+        "status": status.value,
+        "reason_code": reason_code,
+        "max_candidates": max_candidates,
+        "max_depth": max_depth,
+        "max_steps": max_steps,
+        "max_parallelism": max_parallelism,
+        "evidence_spans": [_span_payload(span) for span in gate.evidence_spans],
+    }
+    return ComputeBudgetDecision(
+        budget_id=_digest(payload),
+        policy_version="compute_budget.v1",
+        gate_decision_id=gate.decision_id,
+        gate_policy_version=gate.policy_version,
+        gate_input_digest=gate.input_digest,
+        status=status,
+        reason_code=reason_code,
+        max_candidates=max_candidates,
+        max_depth=max_depth,
+        max_steps=max_steps,
+        max_wallclock_ms=None,
+        max_parallelism=max_parallelism,
+        evidence_spans=gate.evidence_spans,
+        explanation="budget prose",
+    )
+
+
+def _run(
+    *,
+    residual: ContractResidual,
+    gate: SearchGateDecision,
+    budget: ComputeBudgetDecision,
+    problem_frame_digest: str = "f" * 64,
+    assessment_id: str = "assessment-a",
+) -> GeometricSearchRun:
+    outcome = initialize_geometric_search_run(
+        problem_frame_digest=problem_frame_digest,
+        contract_assessment_id=assessment_id,
+        residual_ids=(residual.residual_id,),
+        gate_decision=gate,
+        compute_budget=budget,
+        operator_set_id=candidate_operator_set_id(),
+        operator_set_version=CANDIDATE_OPERATOR_SET_VERSION,
+    )
+    assert isinstance(outcome, GeometricSearchRun)
+    return outcome
+
+
+def _cue(
+    *,
+    quantity_mention_id: str = "q1",
+    entity_mention_id: str = "e1",
+    quantity_kind: str = "count",
+    evidence_spans: tuple[SourceSpan, ...] | None = None,
+    unit_mention_id: str | None = None,
+) -> GroundedQuantityEntityCue:
+    return GroundedQuantityEntityCue(
+        quantity_mention_id=quantity_mention_id,
+        entity_mention_id=entity_mention_id,
+        quantity_kind=quantity_kind,
+        evidence_spans=evidence_spans if evidence_spans is not None else (_span(),),
+        unit_mention_id=unit_mention_id,
+    )
+
+
+def _chain() -> tuple[
+    ContractResidual,
+    SearchGateDecision,
+    ComputeBudgetDecision,
+    GeometricSearchRun,
+    GroundedQuantityEntityCue,
+]:
+    residual = _residual()
+    gate = _gate(residual)
+    budget = _budget(gate)
+    run = _run(residual=residual, gate=gate, budget=budget)
+    cue = _cue(evidence_spans=residual.evidence_spans)
+    return residual, gate, budget, run, cue
+
+
+def _build(**changes: object) -> CandidateOperatorResult | CandidateOperatorRefusal:
+    residual, gate, budget, run, cue = _chain()
+    values: dict[str, object] = {
+        "residual": residual,
+        "search_gate": gate,
+        "compute_budget": budget,
+        "run": run,
+        "problem_frame_digest": run.problem_frame_digest,
+        "original_contract_assessment_id": run.contract_assessment_id,
+        "grounded_quantity_entity_cues": (cue,),
+    }
+    values.update(changes)
+    return build_quantity_entity_binding_candidate(**values)  # type: ignore[arg-type]
+
+
+def test_valid_chain_produces_candidate_operator_result() -> None:
+    outcome = _build()
+    assert isinstance(outcome, CandidateOperatorResult)
+    assert outcome.operator_name == QUANTITY_ENTITY_BINDING_CANDIDATE_OPERATOR_NAME
+    assert outcome.operator_version == QUANTITY_ENTITY_BINDING_CANDIDATE_OPERATOR_VERSION
+    assert outcome.candidate_organ == QUANTITY_ENTITY_BINDING_CANDIDATE_ORGAN
+    assert outcome.attempt_index == 0
+    assert outcome.candidate_attempt.budget_charge == BudgetCharge(candidates=1, steps=1)
+    assert outcome.candidate_attempt.depth == 1
+    assert outcome.candidate_attempt.replay_status == CandidateReplayStatus.REPLAY_PENDING
+    assert outcome.candidate_attempt.replay_blockers == ()
+    assert outcome.reason_codes == ()
+
+
+def test_candidate_payload_matches_schema() -> None:
+    outcome = _build()
+    assert isinstance(outcome, CandidateOperatorResult)
+    payload_dict = dict(outcome.candidate_reconstruction.candidate_payload)
+    assert payload_dict == {
+        "binding_type": "quantity_entity",
+        "candidate_organ": QUANTITY_ENTITY_BINDING_CANDIDATE_ORGAN,
+        "entity_mention_id": "e1",
+        "kind": "mention_binding",
+        "quantity_kind": "count",
+        "quantity_mention_id": "q1",
+        "relation_type": "quantity_entity",
+        "source": "GroundedQuantityEntityCue",
+        "unit_mention_id": "",
+    }
+
+
+def test_deterministic_result_id() -> None:
+    first = _build()
+    second = _build()
+    assert isinstance(first, CandidateOperatorResult)
+    assert isinstance(second, CandidateOperatorResult)
+    assert first.operator_result_id == second.operator_result_id
+
+
+def test_deterministic_attempt_id() -> None:
+    first = _build()
+    second = _build()
+    assert isinstance(first, CandidateOperatorResult)
+    assert isinstance(second, CandidateOperatorResult)
+    assert first.attempt_id == second.attempt_id
+
+
+def test_deterministic_candidate_digest() -> None:
+    first = _build()
+    second = _build()
+    assert isinstance(first, CandidateOperatorResult)
+    assert isinstance(second, CandidateOperatorResult)
+    assert first.candidate_digest == second.candidate_digest
+
+
+def test_explanation_does_not_affect_canonical_ids() -> None:
+    first = _build(explanation="hello")
+    second = _build(explanation="world")
+    assert isinstance(first, CandidateOperatorResult)
+    assert isinstance(second, CandidateOperatorResult)
+    assert first.operator_result_id == second.operator_result_id
+    assert first.attempt_id == second.attempt_id
+    assert first.candidate_digest == second.candidate_digest
+
+
+def test_evidence_span_order_preserved() -> None:
+    spans = (_span("A"), _span("B"), _span("C"))
+    residual = _residual(evidence_spans=spans)
+    gate = _gate(residual)
+    budget = _budget(gate)
+    run = _run(residual=residual, gate=gate, budget=budget)
+    cue = _cue(evidence_spans=spans)
+    outcome = _build(residual=residual, search_gate=gate, compute_budget=budget, run=run, grounded_quantity_entity_cues=(cue,))
+    assert isinstance(outcome, CandidateOperatorResult)
+    assert outcome.evidence_spans == spans
+    assert outcome.candidate_attempt.evidence_spans == spans
+
+
+def test_duplicate_evidence_spans_preserved() -> None:
+    spans = (_span("A"), _span("A"))
+    residual = _residual(evidence_spans=spans)
+    gate = _gate(residual)
+    budget = _budget(gate)
+    run = _run(residual=residual, gate=gate, budget=budget)
+    cue = _cue(evidence_spans=spans)
+    outcome = _build(residual=residual, search_gate=gate, compute_budget=budget, run=run, grounded_quantity_entity_cues=(cue,))
+    assert isinstance(outcome, CandidateOperatorResult)
+    assert outcome.evidence_spans == spans
+
+
+def test_run_candidate_attempts_unchanged() -> None:
+    residual, gate, budget, run, cue = _chain()
+    run_attempts_before = tuple(run.candidate_attempts)
+    _build(residual=residual, search_gate=gate, compute_budget=budget, run=run, grounded_quantity_entity_cues=(cue,))
+    assert run.candidate_attempts == run_attempts_before
+
+
+def test_spine_compatibility() -> None:
+    residual, gate, budget, run, cue = _chain()
+    outcome = build_quantity_entity_binding_candidate(
+        residual=residual,
+        search_gate=gate,
+        compute_budget=budget,
+        run=run,
+        problem_frame_digest=run.problem_frame_digest,
+        original_contract_assessment_id=run.contract_assessment_id,
+        grounded_quantity_entity_cues=(cue,),
+    )
+    assert isinstance(outcome, CandidateOperatorResult)
+    bound_attempt = bind_candidate_attempt_to_run(original_run=run, candidate_operator_result=outcome)
+    adapter_input = build_replay_adapter_input_from_binding(
+        run=run,
+        binding=bound_attempt,
+        candidate_operator_result=outcome,
+    )
+    assert isinstance(adapter_input, ReplayAdapterRefusal)
+    assert any("contract_replay_unavailable" in rc or "operator_replay_unavailable" in rc or "unsupported_candidate_organ" in rc for rc in adapter_input.reason_codes)
+
+
+# Refusal tests
+def test_wrong_residual_kind() -> None:
+    residual = _residual(residual_kind=ResidualKind.MISSING_ROLE)
+    gate = _gate(residual, reason_code="eligible_missing_relation")
+    budget = _budget(gate, reason_code="budget_allowed_missing_relation")
+    run = _run(residual=residual, gate=gate, budget=budget)
+    outcome = build_quantity_entity_binding_candidate(
+        residual=residual,
+        search_gate=gate,
+        compute_budget=budget,
+        run=run,
+        problem_frame_digest=run.problem_frame_digest,
+        original_contract_assessment_id=run.contract_assessment_id,
+        grounded_quantity_entity_cues=(_cue(evidence_spans=residual.evidence_spans),),
+    )
+    assert isinstance(outcome, CandidateOperatorRefusal)
+    assert CandidateOperatorRefusalReason.UNSUPPORTED_RESIDUAL_KIND.value in outcome.reason_codes
+
+
+def test_wrong_residual_code() -> None:
+    residual = _residual(residual_code="some_other_code")
+    gate = _gate(residual)
+    budget = _budget(gate)
+    run = _run(residual=residual, gate=gate, budget=budget)
+    outcome = build_quantity_entity_binding_candidate(
+        residual=residual,
+        search_gate=gate,
+        compute_budget=budget,
+        run=run,
+        problem_frame_digest=run.problem_frame_digest,
+        original_contract_assessment_id=run.contract_assessment_id,
+        grounded_quantity_entity_cues=(_cue(evidence_spans=residual.evidence_spans),),
+    )
+    assert isinstance(outcome, CandidateOperatorRefusal)
+    assert CandidateOperatorRefusalReason.UNSUPPORTED_RESIDUAL_CODE.value in outcome.reason_codes
+
+
+def test_unsupported_candidate_organ() -> None:
+    residual = _residual(candidate_organ="some_other_organ")
+    gate = _gate(residual)
+    budget = _budget(gate)
+    run = _run(residual=residual, gate=gate, budget=budget)
+    outcome = build_quantity_entity_binding_candidate(
+        residual=residual,
+        search_gate=gate,
+        compute_budget=budget,
+        run=run,
+        problem_frame_digest=run.problem_frame_digest,
+        original_contract_assessment_id=run.contract_assessment_id,
+        grounded_quantity_entity_cues=(_cue(evidence_spans=residual.evidence_spans),),
+    )
+    assert isinstance(outcome, CandidateOperatorRefusal)
+    assert CandidateOperatorRefusalReason.UNSUPPORTED_CANDIDATE_ORGAN.value in outcome.reason_codes
+
+
+def test_missing_evidence_spans() -> None:
+    cue = GroundedQuantityEntityCue(
+        quantity_mention_id="q1",
+        entity_mention_id="e1",
+        quantity_kind="count",
+        evidence_spans=(),
+    )
+    outcome = _build(grounded_quantity_entity_cues=(cue,))
+    assert isinstance(outcome, CandidateOperatorRefusal)
+    assert CandidateOperatorRefusalReason.MISSING_EVIDENCE_SPANS.value in outcome.reason_codes
+
+
+def test_malformed_evidence_spans() -> None:
+    bad_span = SimpleNamespace(text="bad", start=-1, end=3, sentence_index=0)
+    cue = GroundedQuantityEntityCue(
+        quantity_mention_id="q1",
+        entity_mention_id="e1",
+        quantity_kind="count",
+        evidence_spans=(bad_span,),  # type: ignore[arg-type]
+    )
+    outcome = _build(grounded_quantity_entity_cues=(cue,))
+    assert isinstance(outcome, CandidateOperatorRefusal)
+    assert CandidateOperatorRefusalReason.MALFORMED_EVIDENCE_SPAN.value in outcome.reason_codes
+
+
+def test_ungrounded_quantity_cue() -> None:
+    outcome = _build(grounded_quantity_entity_cues=("not a cue",))
+    assert isinstance(outcome, CandidateOperatorRefusal)
+    assert CandidateOperatorRefusalReason.INVALID_QUANTITY_ENTITY_CUE_TYPE.value in outcome.reason_codes
+
+
+def test_zero_cues() -> None:
+    outcome = _build(grounded_quantity_entity_cues=())
+    assert isinstance(outcome, CandidateOperatorRefusal)
+    assert CandidateOperatorRefusalReason.MISSING_TYPED_CUE.value in outcome.reason_codes
+
+
+def test_multiple_cues() -> None:
+    cue1 = _cue()
+    cue2 = _cue()
+    outcome = _build(grounded_quantity_entity_cues=(cue1, cue2))
+    assert isinstance(outcome, CandidateOperatorRefusal)
+    assert CandidateOperatorRefusalReason.QUANTITY_ENTITY_CUE_COUNT_MISMATCH.value in outcome.reason_codes
+
+
+def test_empty_quantity_mention_id() -> None:
+    outcome = _build(grounded_quantity_entity_cues=(_cue(quantity_mention_id=""),))
+    assert isinstance(outcome, CandidateOperatorRefusal)
+    assert CandidateOperatorRefusalReason.EMPTY_QUANTITY_MENTION_ID.value in outcome.reason_codes
+
+
+def test_empty_entity_mention_id() -> None:
+    outcome = _build(grounded_quantity_entity_cues=(_cue(entity_mention_id=""),))
+    assert isinstance(outcome, CandidateOperatorRefusal)
+    assert CandidateOperatorRefusalReason.EMPTY_ENTITY_MENTION_ID.value in outcome.reason_codes
+
+
+def test_empty_quantity_kind() -> None:
+    outcome = _build(grounded_quantity_entity_cues=(_cue(quantity_kind=""),))
+    assert isinstance(outcome, CandidateOperatorRefusal)
+    assert CandidateOperatorRefusalReason.EMPTY_QUANTITY_KIND.value in outcome.reason_codes
+
+
+def test_empty_unit_mention_id() -> None:
+    outcome = _build(grounded_quantity_entity_cues=(_cue(unit_mention_id=""),))
+    assert isinstance(outcome, CandidateOperatorRefusal)
+    assert CandidateOperatorRefusalReason.EMPTY_UNIT_MENTION_ID.value in outcome.reason_codes
+
+
+def test_state_change_surface_backdoor() -> None:
+    residual = _residual(candidate_organ="unary_delta_transition")
+    gate = _gate(residual)
+    budget = _budget(gate)
+    run = _run(residual=residual, gate=gate, budget=budget)
+    outcome = build_quantity_entity_binding_candidate(
+        residual=residual,
+        search_gate=gate,
+        compute_budget=budget,
+        run=run,
+        problem_frame_digest=run.problem_frame_digest,
+        original_contract_assessment_id=run.contract_assessment_id,
+        grounded_quantity_entity_cues=(_cue(evidence_spans=residual.evidence_spans),),
+    )
+    assert isinstance(outcome, CandidateOperatorRefusal)
+    assert CandidateOperatorRefusalReason.UNSUPPORTED_CANDIDATE_ORGAN.value in outcome.reason_codes
+
+
+def test_attempt_index_not_zero() -> None:
+    outcome = _build(attempt_index=1)
+    assert isinstance(outcome, CandidateOperatorRefusal)
+    assert CandidateOperatorRefusalReason.ATTEMPT_INDEX_EXCEEDS_BUDGET.value in outcome.reason_codes
+
+
+def test_residual_id_not_in_run() -> None:
+    residual = _residual()
+    gate = _gate(residual)
+    budget = _budget(gate)
+    run = _run(residual=residual, gate=gate, budget=budget)
+    other_residual = _residual(residual_id="other", evidence_spans=residual.evidence_spans)
+    outcome = build_quantity_entity_binding_candidate(
+        residual=other_residual,
+        search_gate=gate,
+        compute_budget=budget,
+        run=run,
+        problem_frame_digest=run.problem_frame_digest,
+        original_contract_assessment_id=run.contract_assessment_id,
+        grounded_quantity_entity_cues=(_cue(evidence_spans=residual.evidence_spans),),
+    )
+    assert isinstance(outcome, CandidateOperatorRefusal)
+    assert CandidateOperatorRefusalReason.INVALID_OPERATOR_INPUT.value in outcome.reason_codes
+
+
+def test_problem_frame_mismatch() -> None:
+    outcome = _build(problem_frame_digest="wrong" * 8)
+    assert isinstance(outcome, CandidateOperatorRefusal)
+    assert CandidateOperatorRefusalReason.INVALID_OPERATOR_INPUT.value in outcome.reason_codes
+
+
+def test_contract_assessment_mismatch() -> None:
+    outcome = _build(original_contract_assessment_id="wrong_assessment")
+    assert isinstance(outcome, CandidateOperatorRefusal)
+    assert CandidateOperatorRefusalReason.INVALID_OPERATOR_INPUT.value in outcome.reason_codes
+
+
+def test_search_gate_not_eligible() -> None:
+    residual, valid_gate, valid_budget, run, cue = _chain()
+    gate = _gate(residual, status=SearchGateStatus.INELIGIBLE)
+    outcome = build_quantity_entity_binding_candidate(
+        residual=residual,
+        search_gate=gate,
+        compute_budget=valid_budget,
+        run=run,
+        problem_frame_digest=run.problem_frame_digest,
+        original_contract_assessment_id=run.contract_assessment_id,
+        grounded_quantity_entity_cues=(_cue(evidence_spans=residual.evidence_spans),),
+    )
+    assert isinstance(outcome, CandidateOperatorRefusal)
+    assert CandidateOperatorRefusalReason.INELIGIBLE_SEARCH_GATE.value in outcome.reason_codes
+
+
+def test_compute_budget_not_allowed() -> None:
+    residual, gate, valid_budget, run, cue = _chain()
+    budget = _budget(gate, status=ComputeBudgetStatus.BUDGET_BLOCKED)
+    outcome = build_quantity_entity_binding_candidate(
+        residual=residual,
+        search_gate=gate,
+        compute_budget=budget,
+        run=run,
+        problem_frame_digest=run.problem_frame_digest,
+        original_contract_assessment_id=run.contract_assessment_id,
+        grounded_quantity_entity_cues=(_cue(evidence_spans=residual.evidence_spans),),
+    )
+    assert isinstance(outcome, CandidateOperatorRefusal)
+    assert CandidateOperatorRefusalReason.NON_ALLOWED_BUDGET.value in outcome.reason_codes
diff --git a/tests/test_replay_adapter_bound_attempt.py b/tests/test_replay_adapter_bound_attempt.py
index 3e0ec37b..f936e7e9 100644
--- a/tests/test_replay_adapter_bound_attempt.py
+++ b/tests/test_replay_adapter_bound_attempt.py
@@ -135,7 +135,7 @@ def _chain() -> tuple[GeometricSearchRun, object, CandidateAttemptRunBinding]:
         gate_decision=gate,
         compute_budget=budget,
         operator_set_id=candidate_operator_set_id(),
-        operator_set_version="candidate_operators.v1",
+        operator_set_version="candidate_operators.v2",
     )
     assert isinstance(run, GeometricSearchRun)
     result = build_missing_role_candidate(
diff --git a/tests/test_run_attempt_binding.py b/tests/test_run_attempt_binding.py
index 53617227..211b067f 100644
--- a/tests/test_run_attempt_binding.py
+++ b/tests/test_run_attempt_binding.py
@@ -131,7 +131,7 @@ def _chain() -> tuple[ContractResidual, GeometricSearchRun, object]:
         gate_decision=gate,
         compute_budget=budget,
         operator_set_id=candidate_operator_set_id(),
-        operator_set_version="candidate_operators.v1",
+        operator_set_version="candidate_operators.v2",
     )
     assert isinstance(run, GeometricSearchRun)
     result = build_missing_role_candidate(
diff --git a/tests/test_sealed_practice_trace_bound_episode.py b/tests/test_sealed_practice_trace_bound_episode.py
index 1f7c9f71..acdb6b68 100644
--- a/tests/test_sealed_practice_trace_bound_episode.py
+++ b/tests/test_sealed_practice_trace_bound_episode.py
@@ -151,7 +151,7 @@ def _chain() -> tuple[
         gate_decision=gate,
         compute_budget=budget,
         operator_set_id=candidate_operator_set_id(),
-        operator_set_version="candidate_operators.v1",
+        operator_set_version="candidate_operators.v2",
     )
     assert isinstance(run, GeometricSearchRun)
     result = build_missing_role_candidate(