126 lines
4.9 KiB
Python
126 lines
4.9 KiB
Python
"""Agent OS consensus engine runtime."""
|
|
|
|
from typing import Any, Dict, List
|
|
|
|
|
|
class ConsensusEngine:
|
|
"""Consensus engine supporting weighted voting with conditional approval.
|
|
|
|
Decision outcomes:
|
|
- "approve": approved outright
|
|
- "reject": explicitly rejected
|
|
- "review": needs review (tie or insufficient votes)
|
|
- "conditional_approval": approved pending further review (new in V1R1C5)
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.base_weights = {
|
|
"custom_gpt": 2,
|
|
"windows_codex": 2,
|
|
"server_agent": 1,
|
|
}
|
|
# New in V1R1C5: quality score, historical reliability, domain expertise
|
|
self.quality_score_factors = {
|
|
"code_quality": 0.4,
|
|
"test_coverage": 0.3,
|
|
"response_time": 0.3,
|
|
}
|
|
self.historical_reliability_factors = {
|
|
"recent_acceptance_rate": 0.5,
|
|
"longevity_score": 0.3,
|
|
"error_rate": 0.2,
|
|
}
|
|
self.domain_expertise_factors = {
|
|
"specialization_match": 0.5,
|
|
"experience_years": 0.3,
|
|
"certifications": 0.2,
|
|
}
|
|
|
|
def calculate_quality_score(self, agent_id: str, agent_type: str) -> float:
|
|
"""Calculate quality score for an agent based on its performance metrics."""
|
|
# Placeholder implementation; in production, fetch metrics from telemetry
|
|
return 0.85 # Default quality score
|
|
|
|
def calculate_historical_reliability(self, agent_id: str, agent_type: str) -> float:
|
|
"""Calculate historical reliability score for an agent."""
|
|
# Placeholder implementation; in production, fetch from store
|
|
return 0.90 # Default reliability score
|
|
|
|
def calculate_domain_expertise(self, agent_id: str, agent_type: str) -> float:
|
|
"""Calculate domain expertise score for an agent."""
|
|
# Placeholder implementation; in production, fetch from agent registry
|
|
return 0.75 # Default expertise score
|
|
|
|
def calculate(self, votes: List[dict], use_dynamic_weights: bool = False) -> dict:
|
|
result = {"approve": 0, "reject": 0}
|
|
for vote in votes:
|
|
base_weight = self.base_weights.get(vote.get("agent_type"), 1)
|
|
if use_dynamic_weights:
|
|
quality = self.calculate_quality_score(vote.get("agent_id", ""), vote.get("agent_type", ""))
|
|
reliability = self.calculate_historical_reliability(vote.get("agent_id", ""), vote.get("agent_type", ""))
|
|
expertise = self.calculate_domain_expertise(vote.get("agent_id", ""), vote.get("agent_type", ""))
|
|
weight = base_weight * quality * reliability * expertise
|
|
else:
|
|
weight = base_weight
|
|
result[vote.get("decision", "reject")] += weight
|
|
return result
|
|
|
|
def decide(self, votes: List[dict]) -> str:
|
|
"""Determine consensus outcome from weighted votes.
|
|
|
|
Returns one of: "approve", "reject", "review", "conditional_approval".
|
|
"""
|
|
result = self.calculate(votes, use_dynamic_weights=True)
|
|
if result["approve"] > result["reject"]:
|
|
# Check if any participant voted with conditions
|
|
conditional_votes = [
|
|
v for v in votes if v.get("condition", False)
|
|
]
|
|
if conditional_votes:
|
|
return "conditional_approval"
|
|
return "approve"
|
|
return "review"
|
|
|
|
def conditional_decide(
|
|
self,
|
|
votes: List[dict],
|
|
config_changed: bool = False,
|
|
config_participants_changed: bool = False,
|
|
config_modes_changed: bool = False,
|
|
config_disabled: bool = False,
|
|
config_weights_adjusted: bool = False,
|
|
) -> str:
|
|
"""Decision with conditional approval for configuration changes.
|
|
|
|
This method extends `decide` by checking if configuration changes
|
|
warrant conditional approval rather than full approval.
|
|
|
|
Conditional approval triggers:
|
|
1. participants list modified (V1R1C5)
|
|
2. modes list modified (V1R1C5)
|
|
3. consensus disabled (V1R1C5)
|
|
4. weights adjusted beyond threshold (V1R1C5)
|
|
"""
|
|
base_decision = self.decide(votes)
|
|
|
|
# If configuration shows major changes, override to conditional_approval
|
|
if config_disabled:
|
|
# Disabling consensus is a critical change
|
|
if base_decision == "approve":
|
|
return "conditional_approval"
|
|
return base_decision
|
|
|
|
if config_participants_changed or config_modes_changed:
|
|
# Participants or modes changed — treat as conditional
|
|
if base_decision == "approve":
|
|
return "conditional_approval"
|
|
return base_decision
|
|
|
|
if config_weights_adjusted:
|
|
# Weights adjusted beyond threshold — treat as conditional
|
|
if base_decision == "approve":
|
|
return "conditional_approval"
|
|
return base_decision
|
|
|
|
return base_decision
|