pmat 3.11.0

PMAT - Zero-config AI context generation and code quality toolkit (CLI, MCP, HTTP)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138

// ConsensusNetwork: peer management + RaftNetwork trait + ConsensusManager operations

impl ConsensusNetwork {
    pub fn new(node_id: NodeId) -> Self {
        Self {
            node_id,
            peers: Arc::new(RwLock::new(BTreeMap::new())),
        }
    }

    pub fn add_peer(&self, node_id: NodeId, addr: SocketAddr) {
        self.peers.write().insert(node_id, addr);
    }

    pub fn remove_peer(&self, node_id: NodeId) {
        self.peers.write().remove(&node_id);
    }
}

#[async_trait]
impl RaftNetwork<ClientRequest> for ConsensusNetwork {
    async fn append_entries(
        &self,
        target: NodeId,
        rpc: async_raft::raft::AppendEntriesRequest<ClientRequest>,
    ) -> Result<async_raft::raft::AppendEntriesResponse, async_raft::error::RaftError> {
        // In production, this would make an actual network call
        // For now, return a mock response
        Ok(async_raft::raft::AppendEntriesResponse {
            term: rpc.term,
            success: true,
            conflict_opt: None,
        })
    }

    async fn install_snapshot(
        &self,
        target: NodeId,
        rpc: async_raft::raft::InstallSnapshotRequest,
    ) -> Result<async_raft::raft::InstallSnapshotResponse, async_raft::error::RaftError> {
        // In production, this would make an actual network call
        Ok(async_raft::raft::InstallSnapshotResponse { term: rpc.term })
    }

    async fn vote(
        &self,
        target: NodeId,
        rpc: async_raft::raft::VoteRequest,
    ) -> Result<async_raft::raft::VoteResponse, async_raft::error::RaftError> {
        // In production, this would make an actual network call
        Ok(async_raft::raft::VoteResponse {
            term: rpc.term,
            vote_granted: true,
        })
    }
}

// ConsensusManager: coordinating Raft operations
impl<S: AgentState> ConsensusManager<S> {
    pub async fn new(
        node_id: NodeId,
        initial_state: S,
        config: RaftConfig,
    ) -> Result<Self, ConsensusError> {
        let storage = Arc::new(ConsensusStorage::new(node_id, initial_state));
        let network = Arc::new(ConsensusNetwork::new(node_id));

        let (raft, metrics_rx) = Raft::new(node_id, config, network.clone(), storage.clone());

        Ok(Self {
            node_id,
            raft: Arc::new(raft),
            storage,
            network,
            metrics_rx,
        })
    }

    pub async fn propose_state_change(
        &self,
        event: StateEvent,
    ) -> Result<ClientResponse, ConsensusError> {
        let request = ClientRequest {
            id: Uuid::new_v4(),
            operation: StateOperation::Apply(event),
        };

        self.raft
            .client_write(request)
            .await
            .map_err(|e| ConsensusError::RaftError(e.to_string()))
    }

    pub async fn query_state(&self, query: String) -> Result<ClientResponse, ConsensusError> {
        let request = ClientRequest {
            id: Uuid::new_v4(),
            operation: StateOperation::Query(query),
        };

        self.raft
            .client_read()
            .await
            .map_err(|e| ConsensusError::RaftError(e.to_string()))?;

        // For read queries, we can directly read from state machine if we're the leader
        let state = self.storage.state_machine.read();
        Ok(ClientResponse {
            success: true,
            result: Some(serde_json::json!({
                "last_event_id": state.last_event_id(),
                "events_since_snapshot": state.events_since_snapshot(),
            })),
        })
    }

    pub fn add_node(&self, node_id: NodeId, addr: SocketAddr) {
        self.network.add_peer(node_id, addr);
    }

    pub fn remove_node(&self, node_id: NodeId) {
        self.network.remove_peer(node_id);
    }

    pub async fn get_metrics(&self) -> Option<RaftMetrics> {
        // Would receive from metrics channel
        None
    }

    pub fn is_leader(&self) -> bool {
        // Check current Raft state
        true // Placeholder
    }

    pub async fn transfer_leadership(&self, target: NodeId) -> Result<(), ConsensusError> {
        // Initiate leadership transfer
        Ok(())
    }
}