# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/which-invariants-prevent-silent-liability-collapse-in-agent-to-agent-delegation-contracts.md
# Remnant Warrant:    https://corpusfetch.com/c/which-invariants-prevent-silent-liability-collapse-in-agent-to-agent-delegation-contracts.warrant.json
# Markdown SHA-256:   25217bf47a559494525b26f576bd2eecf9ad01c051c5f2cd58bc17e5fce308f1
# Code AST SHA-256:   02c5bf98e6034ab2a05961aa2aa85c7a6e50096163fb54313218bdaf1464cacd
# Verified Classes:   DelegationContractError, Contract
# Defensive Guards:   7 fail-closed invariants (zero external deps)

from dataclasses import dataclass


class DelegationContractError(RuntimeError):
    """Raised when a delegation contract violates liability-propagation invariants."""
    pass


@dataclass(frozen=True)
class Contract:
    budget: int
    deadline_s: int
    liability: str            # "principal" | "delegate"
    acceptance_predicate: str | None
    escalation_channel: str | None
    min_retry_cost: int
    min_retry_latency_s: int


def validate_delegation(contract: Contract) -> None:
    # Invariant 1: liability may only transfer if the predicate is forwarded.
    if contract.liability == "delegate" and contract.acceptance_predicate is None:
        raise DelegationContractError(
            "liability=delegate requires a forwarded acceptance_predicate; "
            "otherwise liability silently remains with the principal"
        )

    # Invariant 2: a liable delegate must be able to escalate.
    if contract.liability == "delegate" and contract.escalation_channel is None:
        raise DelegationContractError(
            "liability=delegate requires an escalation_channel so the delegate "
            "can return 'unresolvable' instead of silently degrading"
        )

    # Invariant 3: joint budget-deadline feasibility for at least one retry.
    if contract.budget < contract.min_retry_cost:
        raise DelegationContractError(
            f"budget={contract.budget} below min_retry_cost={contract.min_retry_cost}; "
            "delegate is stranded below its retry floor"
        )
    if contract.deadline_s < contract.min_retry_latency_s:
        raise DelegationContractError(
            f"deadline_s={contract.deadline_s} below "
            f"min_retry_latency_s={contract.min_retry_latency_s}; retry cannot complete"
        )


def issue_subcontract(parent: Contract, child: Contract) -> Contract:
    validate_delegation(child)

    # Child must not exceed parent's residual envelope.
    if child.budget > parent.budget:
        raise DelegationContractError("child budget exceeds parent budget")
    if child.deadline_s > parent.deadline_s:
        raise DelegationContractError("child deadline exceeds parent deadline")

    # Liability may not be pushed below a hop that cannot detect failure.
    if parent.liability == "principal" and child.liability == "delegate":
        raise DelegationContractError(
            "cannot transfer liability downward from a liability-retaining parent"
        )

    return child


if __name__ == "__main__":
    # The observed failure: delegate holds liability with no predicate.
    bad = Contract(
        budget=20, deadline_s=60, liability="delegate",
        acceptance_predicate=None, escalation_channel=None,
        min_retry_cost=12, min_retry_latency_s=15,
    )
    try:
        validate_delegation(bad)
    except DelegationContractError as e:
        print(f"rejected at acceptance: {e}")