# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-an-instance-refuse-a-delegated-capability-whose-policy-scope-is-satisfiable-only-after-the-delegating-organization-has-already-mutated-its-o.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-instance-refuse-a-delegated-capability-whose-policy-scope-is-satisfiable-only-after-the-delegating-organization-has-already-mutated-its-o.warrant.json
# Markdown SHA-256:   8349f1516e969f0d2cf573e0915abb2e61c8c1a0d56ef9e8f9f8099e8f75ee2a
# Code AST SHA-256:   96d20b53a4b6b36c88d7a8eb97be723d18260ffd1f59dcf976524b122d24c7ea
# Verified Classes:   AuditOrderingUnsatisfiable
# Defensive Guards:   2 fail-closed invariants (zero external deps)

class AuditOrderingUnsatisfiable(RuntimeError):
    """Raised when a delegated capability's policy scope reads the delegator's
    audit trail and cannot be satisfied without a delegator-side mutation whose
    completion is not observable to the delegatee. Distinct from Denied: the
    policy was not evaluable, not failed."""
    def __init__(self, capability_id: str, delegator_org: str,
                 audit_high_water_mark: int | None, reason: str):
        self.capability_id = capability_id
        self.delegator_org = delegator_org
        self.audit_high_water_mark = audit_high_water_mark
        self.reason = reason
        super().__init__(
            f"capability {capability_id} from {delegator_org} not evaluable: "
            f"{reason} (delegator hwm={audit_high_water_mark})"
        )


def evaluate_delegated_capability(cap, subject, delegator_view):
    # delegator_view exposes: audit_replica, hwm, predicate_scans_partition
    predicate = cap.policy_scope
    if predicate.reads_audit_partition == delegator_view.evaluation_log_partition:
        # Self-referential: evaluation events pollute the scanned partition.
        raise AuditOrderingUnsatisfiable(
            cap.id, cap.delegator_org,
            delegator_view.hwm,
            "evaluation log shares partition with policy-scanned audit trail; "
            "retry cannot converge under strict ordering",
        )
    if not delegator_view.exposes_read_your_writes_to_peers():
        # Contract does not guarantee external visibility; transient is legitimate.
        return predicate.evaluate_with_backoff(delegator_view.audit_replica)
    result = predicate.evaluate_once(delegator_view.audit_replica)
    if result.satisfied:
        return result
    if result.missing_inputs_are_delegator_mutations():
        # Not evaluable; do not retry, do not deny.
        raise AuditOrderingUnsatisfiable(
            cap.id, cap.delegator_org,
            delegator_view.hwm,
            f"predicate requires delegator mutation visible at hwm>{delegator_view.hwm}",
        )
    return result  # genuine denial