BFT Consensus & Distributed Ledger

class PBFTReplica:
    """Three-phase commit with Ed25519-signed messages."""

    async def handle_request(self, request: ClientRequest):
        if not self.is_primary:
            return self.forward_to_primary(request)

        seq = self.next_sequence()
        digest = sha256(serialize(request)).hexdigest()

        pre_prepare = PBFTMessage(
            msg_type=MessageType.PRE_PREPARE,
            view=self.current_view,
            sequence=seq,
            digest=digest,
            sender_id=self.replica_id,
            signature=self.sign(f"AEOS/pbft/pp/{self.current_view}/{seq}/{digest}")
        )
        await self.broadcast(pre_prepare)

    async def handle_prepare(self, msg: PBFTMessage):
        self.prepare_log[msg.sequence].add(msg)
        if len(self.prepare_log[msg.sequence]) >= self.quorum_size:
            self.phase[msg.sequence] = Phase.PREPARED
            commit = self._build_commit(msg.view, msg.sequence, msg.digest)
            await self.broadcast(commit)

    async def handle_commit(self, msg: PBFTMessage):
        self.commit_log[msg.sequence].add(msg)
        if len(self.commit_log[msg.sequence]) >= self.quorum_size:
            self.phase[msg.sequence] = Phase.COMMITTED
            result = self.execute(msg.sequence)
            await self.reply_to_client(result)

async def initiate_view_change(self):
    """Trigger view change when primary is suspected faulty."""
    self.current_view += 1
    new_primary = self.current_view % self.network_size

    # Collect prepared certificates to carry forward
    prepared_certs = {
        seq: cert for seq, cert in self.certificates.items()
        if cert.msg_type == MessageType.PREPARE
    }

    view_change = PBFTMessage(
        msg_type=MessageType.VIEW_CHANGE,
        view=self.current_view,
        sequence=self.last_stable_checkpoint,
        digest=serialize_certs(prepared_certs),
        sender_id=self.replica_id,
        signature=self.sign(
            f"AEOS/pbft/vc/{self.current_view}/{self.last_stable_checkpoint}"
        )
    )
    await self.broadcast(view_change)

async def handle_new_view(self, msg: PBFTMessage):
    """New primary broadcasts NEW_VIEW after collecting 2f+1 VIEW_CHANGE."""
    if not self.verify_new_view_proof(msg):
        return  # invalid proof

    self.current_view = msg.view
    # Re-propose any operations that were prepared but not committed
    for cert in msg.prepared_certificates:
        await self.handle_request(cert.original_request)

class Phase(IntEnum):
    """Monotonic execution phase per sequence slot."""
    IDLE         = 0  # no request assigned
    PRE_PREPARED = 1  # primary broadcast PRE_PREPARE
    PREPARED     = 2  # received 2f+1 matching PREPARE
    COMMITTED    = 3  # received 2f+1 matching COMMIT
    EXECUTED     = 4  # operation applied to state machine

    def can_advance_to(self, target: "Phase") -> bool:
        """Phases advance monotonically within a view."""
        return target.value == self.value + 1

class DistributedLedger:
    """Replicated economic state machine over PBFTNetwork."""

    def __init__(self, network: PBFTNetwork):
        self.network = network
        self.seen_ops: set[str] = set()
        self.state: dict[str, Any] = {}

    async def submit_operation(
        self, op_type: str, payload: dict
    ) -> ConsensusResult:
        """Submit with deduplication via content-addressed hash."""
        op_hash = sha256(
            f"{op_type}:{serialize(payload)}".encode()
        ).hexdigest()

        if op_hash in self.seen_ops:
            return ConsensusResult(status="duplicate", seq=-1)

        self.seen_ops.add(op_hash)
        result = await self.network.submit(
            operation={"type": op_type, "payload": payload, "hash": op_hash}
        )

        if result.committed:
            self._apply(op_type, payload)
        return result

    def _apply(self, op_type: str, payload: dict):
        """Deterministic state transition after consensus."""
        match op_type:
            case "agent_registered":
                self.state[payload["did"]] = payload
            case "contract_created":
                self.state[payload["contract_id"]] = payload
            case "obligation_fulfilled":
                contract = self.state[payload["contract_id"]]
                contract["milestones_complete"] += 1
            case "dispute_filed":
                self.state[payload["dispute_id"]] = payload