# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/in-what-scenarios-must-an-instance-reject-a-peer-spiffe-trust-bundle-refresh-that-is-cryptographically-valid-but-semantically-stale.md
# Remnant Warrant:    https://corpusfetch.com/c/in-what-scenarios-must-an-instance-reject-a-peer-spiffe-trust-bundle-refresh-that-is-cryptographically-valid-but-semantically-stale.warrant.json
# Markdown SHA-256:   a5806049e90936e2d2f06a7310e5b545fd945ccc37b1e301c167860fb531489a
# Code AST SHA-256:   541b869e4200113e5e768ebc0fe0a653d8ab7a2e8a12b27750d063cc82ae06da
# Verified Classes:   StaleTrustBundleError, TrustBundle
# Defensive Guards:   7 fail-closed invariants (zero external deps)

import hashlib
from dataclasses import dataclass
from datetime import datetime, timedelta


class StaleTrustBundleError(RuntimeError):
    """Raised when a cryptographically valid bundle is semantically stale."""


@dataclass(frozen=True)
class TrustBundle:
    sequence: int
    generated_at: datetime
    authority_ids: frozenset
    path_constraints: dict
    signature_valid: bool

    def content_hash(self) -> str:
        joined = "|".join(sorted(self.authority_ids))
        return hashlib.sha256(joined.encode()).hexdigest()


def apply_bundle_refresh(
    held: TrustBundle,
    incoming: TrustBundle,
    now: datetime,
    max_bundle_age: timedelta,
    revocation_grace: timedelta,
    revocation_observed_at: datetime | None,
) -> TrustBundle:
    # Cryptographic validity is necessary but never sufficient.
    if not incoming.signature_valid:
        raise StaleTrustBundleError("incoming bundle signature invalid")

    # (1) Monotonic sequence; equality with divergent content is a fork.
    if incoming.sequence < held.sequence:
        raise StaleTrustBundleError(
            f"sequence regression {incoming.sequence} < {held.sequence}"
        )
    if (
        incoming.sequence == held.sequence
        and incoming.content_hash() != held.content_hash()
    ):
        raise StaleTrustBundleError("same sequence, divergent content (fork or lying cache)")

    # (2) Age judged from the signed generation timestamp, not transport headers.
    if now - incoming.generated_at > max_bundle_age:
        raise StaleTrustBundleError(
            f"bundle generation age {now - incoming.generated_at} exceeds {max_bundle_age}"
        )

    # (3) Removals require positive revocation evidence within grace.
    removed = held.authority_ids - incoming.authority_ids
    if removed:
        if revocation_observed_at is None:
            raise StaleTrustBundleError(
                f"authority removal {sorted(removed)} without revocation signal"
            )
        if now - revocation_observed_at > revocation_grace:
            raise StaleTrustBundleError(
                f"revocation grace {revocation_grace} exceeded for {sorted(removed)}"
            )

    # (4) Path-constraint widening is a trust-topology mutation.
    for authority_id, new_scope in incoming.path_constraints.items():
        old_scope = held.path_constraints.get(authority_id)
        if old_scope is not None and not _is_narrowing_or_equal(old_scope, new_scope):
            raise StaleTrustBundleError(
                f"path constraint widened for {authority_id}: {old_scope!r} -> {new_scope!r}"
            )

    return incoming


def _is_narrowing_or_equal(old_scope: str, new_scope: str) -> bool:
    # Simplified prefix containment: new_scope must be a prefix of old_scope.
    return old_scope.startswith(new_scope)