""" QualityEngine - centralized scoring, domain-term whitelist, and budget enforcement. Usage: engine = QualityEngine() report = engine.score_output("Some agent output text", "product_owner") # report is a QualityReport with ARI scores, budget check, dimensions """ import re import textstat from .schemas import ARIResult, QualityDimension, QualityReport class DomainTermWhitelist: """Known technical terms that ARI should not flag as verbosity. When ARI (Automated Readability Index) scores technical text, domain-specific multi-syllable terms like "microservices" and "containerization" inflate the score. This class replaces those terms with short placeholders so the ARI measures structural complexity rather than vocabulary density. """ def __init__(self): self._terms: set[str] = set() self._load_defaults() def _load_defaults(self): """Seed from 100+ software engineering terms.""" self._terms = { # Architecture patterns "microservices", "service-oriented", "event-driven", "domain-driven", "clean architecture", "hexagonal", "layered architecture", "command query responsibility segregation", "cqrs", "publish-subscribe", "request-response", "point-to-point", # DDD terms "bounded context", "ubiquitous language", "aggregate root", "domain event", "value object", "domain service", "anti-corruption layer", "repository pattern", "factory method", # Database/data terms "eventual consistency", "strong consistency", "relational database", "non-relational", "denormalization", "materialized view", "connection pooling", "sharding", "replication", "acid", "base", "cap theorem", "crud", # Infrastructure "continuous integration", "continuous deployment", "infrastructure as code", "blue-green deployment", "canary deployment", "circuit breaker", "bulkhead", # Security "cross-site scripting", "cross-site request forgery", "sql injection", "denial of service", "man-in-the-middle", "role-based access control", "attribute-based access control", "authentication", "authorization", "single sign-on", # Testing "acceptance criteria", "unit test", "integration test", "end-to-end test", "regression test", "performance test", # General SE "dependency injection", "inversion of control", "separation of concerns", "single responsibility", "open-closed principle", "liskov substitution", "interface segregation", "dependency inversion", # Additional technical terms "asynchronous", "containerization", "orchestrator", "idempotency", "idempotent", "serialization", "deserialization", "denormalized", "polyglot", "middleware", "websocket", "webhook", # More architecture "event sourcing", "saga pattern", "strangler fig", "sidecar", "ambassador", "adapter", "proxy", # More data "data warehouse", "data lake", "oltp", "olap", "etl", # More infrastructure "horizontal scaling", "vertical scaling", "auto-scaling", "load balancing", "service mesh", "api gateway", # More security "oauth", "jwt", "saml", "openid connect", "zero trust", "defense in depth", "principle of least privilege", } def strip_known_terms(self, text: str) -> str: """Replace whitelisted terms with short tokens for ARI scoring. Multi-word terms are matched longest-first to avoid partial replacement (e.g., "blue-green deployment" replaces before "deployment" alone). """ result = text for term in sorted(self._terms, key=len, reverse=True): result = re.sub( r"\b" + re.escape(term) + r"\b", " word ", result, flags=re.IGNORECASE, ) # Clean up extra spaces from placeholder insertion result = re.sub(r" +", " ", result) return result.strip() def get_matched_terms(self, text: str) -> list[str]: """Return whitelisted terms found in text (for logging/metrics).""" text_lower = text.lower() matched: list[str] = [] for term in sorted(self._terms, key=len, reverse=True): if term.lower() in text_lower: matched.append(term) text_lower = text_lower.replace(term.lower(), "") return matched class QualityEngine: """Centralized ARI scoring, whitelist filtering, and budget enforcement. Each agent role has a calibrated ARI budget derived from research data. The engine computes ARI (with and without domain-term whitelist) plus complementary readability metrics (Flesch-Kincaid, Gunning Fog, Coleman-Liau). Thread-safe: all computation is CPU-bound in textstat, runs synchronously. Use ``score_output_async()`` for async callers. """ CALIBRATED_ARI_BUDGETS: dict[str, float] = { "project_refiner": 12, "product_owner": 12, "business_analyst": 14, "solution_architect": 20, "data_architect": 18, "security_analyst": 16, "ux_designer": 14, "api_designer": 14, "qa_strategist": 16, "devops_architect": 14, "spec_coordinator": 14, "technical_writer": 14, "judge": 18, } def __init__(self, budgets: dict[str, float] | None = None): self._whitelist: DomainTermWhitelist = DomainTermWhitelist() self.budgets = budgets or dict(self.CALIBRATED_ARI_BUDGETS) def score_output(self, text: str, role: str) -> QualityReport: """Score a single agent output. Synchronous - call via ``asyncio.to_thread``. Steps: 1. Compute raw ARI on original text 2. Apply domain-term whitelist (strip known terms) 3. Compute whitelist-adjusted ARI + complementary metrics 4. Check role-specific budget 5. Build and return a complete QualityReport """ if not text.strip(): budget = self.budgets.get(role, 14) return QualityReport( role=role, agent_output="", ari_before_any=0, final_ari=0, ari_result=ARIResult( raw_score=0, whitelist_score=0, budget=budget, passed=True ), word_count=0, sentence_count=0, readability_label="Empty", final_disposition="passed", ) # Compute ARI on raw text (before whitelist) raw_ari = textstat.automated_readability_index(text) # Apply whitelist cleaned = self._whitelist.strip_known_terms(text) matched_terms = self._whitelist.get_matched_terms(text) # Compute whitelist-adjusted ARI adjusted_ari = textstat.automated_readability_index(cleaned) # Compute complementary metrics (on cleaned text) fk_grade = textstat.flesch_kincaid_grade(cleaned) gf_index = textstat.gunning_fog(cleaned) cl_index = textstat.coleman_liau_index(cleaned) syl_per_word = textstat.avg_syllables_per_word(cleaned) word_count = textstat.lexicon_count(cleaned) sent_count = textstat.sentence_count(cleaned) # Budget check budget = self.budgets.get(role, 14) ari_passed = adjusted_ari <= budget # Dimensions dimensions = [ QualityDimension( name="ari_budget", score=round(adjusted_ari, 2), threshold=budget, passed=ari_passed, details=f"ARI {adjusted_ari:.1f} vs budget {budget}", ), QualityDimension( name="readability_fk", score=round(fk_grade, 2), threshold=budget + 2, passed=fk_grade <= budget + 2, details=f"Flesch-Kincaid {fk_grade:.1f}", ), ] ari_result = ARIResult( raw_score=round(raw_ari, 2), whitelist_score=round(adjusted_ari, 2), budget=budget, passed=ari_passed, terms_whitelisted=matched_terms, ) report = QualityReport( role=role, agent_output=text, ari_before_any=round(raw_ari, 2), final_ari=round(adjusted_ari, 2), flesch_kincaid_grade=round(fk_grade, 2), gunning_fog_index=round(gf_index, 2), coleman_liau_index=round(cl_index, 2), avg_syllables_per_word=round(syl_per_word, 2), word_count=word_count, sentence_count=sent_count, readability_label=self._readability_label(adjusted_ari), ari_result=ari_result, dimensions=dimensions, final_disposition="passed" if ari_passed else "budget_exceeded", ) return report async def score_output_async(self, text: str, role: str) -> QualityReport: """Async wrapper - runs ``score_output`` in thread pool. Usage in async orchestrators: report = await self.quality_engine.score_output_async(text, role) """ import asyncio return await asyncio.to_thread(self.score_output, text, role) def detect_stagnation(self, scores: list[float]) -> bool: """Detect stagnation when ARI delta < 0.5 between last two attempts. Returns True if 2+ scores exist and the absolute difference between the last two scores is less than 0.5. """ if len(scores) < 2: return False return abs(scores[-1] - scores[-2]) < 0.5 @staticmethod def _readability_label(ari: float) -> str: """Map ARI score to a readability level label.""" if ari <= 6: return "Elementary" elif ari <= 9: return "Middle School" elif ari <= 12: return "High School" elif ari <= 14: return "College" else: return "Professional"