# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/why-does-an-instance-of-bft-consensus-for-multi-agent-task-commitment-fail-at-the-boundary-between-quorum-certification-and-commitment-finalization.md
# Remnant Warrant:    https://corpusfetch.com/c/why-does-an-instance-of-bft-consensus-for-multi-agent-task-commitment-fail-at-the-boundary-between-quorum-certification-and-commitment-finalization.warrant.json
# Markdown SHA-256:   69a9b3b5a56a089f644f2ff35257801299d497b9782ffa21a9cba80e1fdfaf6f
# Code AST SHA-256:   2a6a0dd762b334b3cf0ff40ca52b8091d05af1aee0199d76f0e78dfdccb5ebc5
# Verified Classes:   CommitmentBoundaryError, TaskCommitmentAgent
# Defensive Guards:   3 fail-closed invariants (zero external deps)

class CommitmentBoundaryError(RuntimeError):
    """Raised when an agent attempts to finalize a task commitment without
    satisfying the QC + lock + no-higher-conflict invariant."""
    pass


class TaskCommitmentAgent:
    def __init__(self, replica_id: int, f: int):
        self.replica_id = replica_id
        self.quorum = 2 * f + 1
        self.lock_view = -1          # highest view for which we hold a QC
        self.lock_slot = None        # logical slot the lock applies to
        self.finalized = {}          # slot -> task assignment

    def receive_qc(self, slot: str, view: int) -> None:
        # Learning a QC is what populates the lock. Signing a proposal does NOT.
        if view > self.lock_view or (view == self.lock_view and slot == self.lock_slot):
            self.lock_view = view
            self.lock_slot = slot

    def finalize(self, slot: str, view: int, task: str, qc_present: bool) -> None:
        if not qc_present:
            raise CommitmentBoundaryError(
                f"replica {self.replica_id}: cannot finalize slot {slot} "
                f"at view {view} without an observed QC"
            )
        if self.lock_slot != slot or self.lock_view < view:
            raise CommitmentBoundaryError(
                f"replica {self.replica_id}: lock state (slot={self.lock_slot}, "
                f"view={self.lock_view}) does not cover finalization of "
                f"slot {slot} at view {view}; QC observed but not locked"
            )
        prior = self.finalized.get(slot)
        if prior is not None and prior != task:
            raise CommitmentBoundaryError(
                f"replica {self.replica_id}: conflicting finalization for slot "
                f"{slot}: already bound to {prior}, refusing {task}"
            )
        self.finalized[slot] = task