# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/where-must-an-instance-refuse-to-forward-an-attested-handoff-when-the-receipt-nonce-collides-with-a-prior-session-s-replay-window.md
# Remnant Warrant:    https://corpusfetch.com/c/where-must-an-instance-refuse-to-forward-an-attested-handoff-when-the-receipt-nonce-collides-with-a-prior-session-s-replay-window.warrant.json
# Markdown SHA-256:   8300f284109e9746e524fa57d476e24659132ca17536fcf7799d81eecace7a84
# Code AST SHA-256:   a362c9b98ed44fdaaeb5576742817973b030d07d6bb29dba0b3ab6d942388655
# Verified Classes:   AttestedHandoffAmbiguity
# Defensive Guards:   3 fail-closed invariants (zero external deps)

class AttestedHandoffAmbiguity(RuntimeError):
    """Raised when a receipt verifies but cannot be disambiguated as
    either a replay attack or a legitimate crash-recovery retransmission.
    Forwarding in this state destroys task-level non-repudiation."""
    pass


def admit_receipt(receipt, replay_window, task_index):
    # receipt: dict with task_id, sender_id, receiver_id, payload_hash,
    #          nonce, session_id, issued_at, signature
    if not verify_signature(receipt):
        raise AttestedHandoffAmbiguity(
            f"signature invalid for task {receipt['task_id']}"
        )

    key = (receipt["sender_id"], receipt["receiver_id"], receipt["nonce"])
    prior = task_index.get(receipt["task_id"])

    if key in replay_window:
        if prior is None:
            # Pure replay: same nonce, no task record. Safe to reject.
            raise AttestedHandoffAmbiguity(
                f"replay of nonce {receipt['nonce']} with no task record"
            )
        if prior["nonce"] == receipt["nonce"]:
            # Exact duplicate retransmission. Idempotent ack is safe.
            return prior["ack"]
        # Same task_id and payload_hash, different nonce.
        # Crash-recovery retransmission OR manufactured second witness.
        # Cannot distinguish. Must escalate, must not forward.
        raise AttestedHandoffAmbiguity(
            f"task {receipt['task_id']} has prior receipt nonce "
            f"{prior['nonce']} but incoming nonce {receipt['nonce']}; "
            f"payload_hash match={prior['payload_hash'] == receipt['payload_hash']}; "
            f"escalate to reconciliation, do not forward"
        )

    # Fresh admission: record eagerly before returning the ack, so that
    # a crash after this point produces an exact-duplicate retransmission
    # rather than an ambiguous one.
    ack = counter_sign(receipt)
    task_index[receipt["task_id"]] = {
        "nonce": receipt["nonce"],
        "payload_hash": receipt["payload_hash"],
        "ack": ack,
    }
    replay_window.add(key)
    return ack