import asyncio from datetime import datetime import json import os import re from typing import Dict, List, Tuple from src.gemini_utils import call_gemini_with_retry, normalize_url from src.logger import log_event from src.mandate_ingestor import get_system_mandates # Memory file to store resolved debates DEBATE_MEMORY_FILE = "src/memory/health_learning.json" async def run_debate_protocol(item: Dict, is_new_link: bool = False) -> Tuple[int, List[str], str, Dict]: """ Executes a Multi-Agent Consensus & Debate Protocol for borderline resources. The panel of experts consists of: 1. Security Architect (Vulnerabilities, Licensing, Supply-Chain) 2. Cloud Native SRE (Production readiness, Scalability, Community activity) 3. AI Platform Engineer (Developer experience, Agentic integrations) Returns: final_score: Resolved impact score (0-100) verified_tags: Consensus tags refined_summary: Refined high-density technical summary debate_data: Logs and details of the debate """ url = item.get("url", "") title = item.get("title", "Unknown Resource") desc = item.get("description", item.get("ai_summary", "")) tags = item.get("tags", []) initial_score = item.get("impact_score", item.get("stars", 3) * 20) # Fallback mapping if stars is used # 0. Check cache using content hash (Recommendation #4) import hashlib raw_content = f"{title}||{desc}||{','.join(sorted(tags))}" content_hash = hashlib.sha256(raw_content.encode("utf-8")).hexdigest() if os.path.exists(DEBATE_MEMORY_FILE): try: with open(DEBATE_MEMORY_FILE, "r") as f: memory_data = json.load(f) cached = memory_data.get("resolved_debates", {}).get(normalize_url(url)) if cached and cached.get("content_hash") == content_hash: log_event(f" [⚖️] CACHE HIT: Skipping debate for '{title}'. Returning cached consensus.") return ( cached["final_consensus_score"], cached.get("final_tags", tags), cached.get("refined_summary", desc), cached ) except Exception as e: log_event(f" [!] Error checking debate cache: {e}") log_event(f" [⚖️] DEBATE TRIGGERED: '{title}' (Initial Score: {initial_score})", section_break=False) # 0. Check if mock mode is requested or required (no keys configured) from src.config import GEMINI_API_KEYS mock_enabled = os.environ.get("MOCK_DEBATE") == "true" or not GEMINI_API_KEYS if mock_enabled: log_event(f" [⚖️] Bypassing Gemini API: Running Offline Mock Debate Simulator...") text_to_scan = (title + " " + desc + " " + " ".join(tags)).lower() # Security Architect if any(x in text_to_scan for x in ["secure", "hardened", "cryptography", "sign", "compliance", "license", "oauth", "token", "vault", "identity", "keycloak", "sops"]): sa_score = 90 sa_just = "Strong security architecture, automated credential isolation, and supply-chain guarantees." else: sa_score = 70 sa_just = "Standard security compliance with typical permission profiles; no specific zero-trust hardening." # Cloud Native SRE if any(x in text_to_scan for x in ["production", "scalable", "monitoring", "prometheus", "ha", "reliability", "redundant", "kubernetes", "operator", "helm", "flux", "argo", "k3s", "draino"]): sre_score = 92 sre_just = "Excellent operational metrics, clear liveness/readiness configuration, and proven recovery behavior." else: sre_score = 65 sre_just = "Acceptable single-instance footprint, but lacks comprehensive scaling runbooks and observability probes." # AI Platform Engineer if any(x in text_to_scan for x in ["agent", "mcp", "llm", "ai", "intelligence", "model", "prompt", "backstage", "developer"]): ai_score = 94 ai_just = "Highly relevant for 2026 cognitive architectures, supporting developer agility and LLM/agent integrations." else: ai_score = 60 ai_just = "Conventional software engineering resource with minimal alignment to agentic orchestration patterns." scores = { "Security Architect": sa_score, "Cloud Native SRE": sre_score, "AI Platform Engineer": ai_score } justifications = { "Security Architect": sa_just, "Cloud Native SRE": sre_just, "AI Platform Engineer": ai_just } for name, score in scores.items(): log_event(f" [>] {name} rated: {score} (Justification: {justifications[name]})") max_score = max(scores.values()) min_score = min(scores.values()) divergence = max_score - min_score debate_transcript = [] if divergence >= 15: log_event(f" [⚖️] Divergence detected ({divergence} points). Starting Mock Debate Round...") scores["Security Architect"] = int((sa_score + 78) / 2) scores["Cloud Native SRE"] = int((sre_score + 80) / 2) scores["AI Platform Engineer"] = int((ai_score + 82) / 2) rebuttals = { "Security Architect": "We must prioritize baseline compliance and permission boundaries even if the developer tool yields high platform speed.", "Cloud Native SRE": "Agreed, but the active community checkins and robust recovery hooks significantly offset the operational risk.", "AI Platform Engineer": "Platform agility is paramount; wrapping this tool in an MCP server exposes its schema for cognitive agent orchestration." } for name, score in scores.items(): debate_transcript.append(f"{name} (Score {score}): {rebuttals[name]}") log_event(f" [>] {name} revised rating to {score}. Rebuttal: {rebuttals[name]}") final_score = int(sum(scores.values()) / len(scores)) log_event(f" [🏁] Consensus Score reached: {final_score}") refined_summary = desc + " — Consensus Audit: The panel aligned on its enterprise maturity, noting its role in streamlining cloud native operations." final_tags = set(tags) if final_score >= 85: final_tags.add("[DE FACTO STANDARD]") if "[COMMUNITY-TOOL]" in final_tags: final_tags.remove("[COMMUNITY-TOOL]") elif final_score >= 70: final_tags.add("[ENTERPRISE-STABLE]") if "[COMMUNITY-TOOL]" in final_tags: final_tags.remove("[COMMUNITY-TOOL]") else: final_tags.add("[COMMUNITY-TOOL]") if "ebpf" in text_to_scan: final_tags.add("[EBPF]") if "wasm" in text_to_scan: final_tags.add("[WASM]") if "gitops" in text_to_scan: final_tags.add("[GITOPS]") if "iac" in text_to_scan: final_tags.add("[IAC]") if any(x in text_to_scan for x in ["agent", "mcp", "ai"]): final_tags.add("[AI]") debate_data = { "url": url, "title": title, "initial_score": initial_score, "final_consensus_score": final_score, "scores": scores, "justifications": justifications, "rebuttals": debate_transcript, "timestamp": datetime.now().isoformat(), "final_tags": sorted(list(final_tags)), "refined_summary": refined_summary, "content_hash": content_hash } try: memory_data = {} if os.path.exists(DEBATE_MEMORY_FILE): try: memory_data = json.load(open(DEBATE_MEMORY_FILE, "r")) except Exception as e: log_event(f"[WARN] load debate memory for mock persist: {str(e)[:100]}") memory_data.setdefault("resolved_debates", {})[normalize_url(url)] = debate_data with open(DEBATE_MEMORY_FILE, "w") as f: json.dump(memory_data, f, indent=2) except Exception as e: log_event(f" [!] Failed to persist debate memory: {e}") return final_score, sorted(list(final_tags)), refined_summary, debate_data system_mandates = get_system_mandates() # Fast-Pass Evaluator call (Recommendation #3) fast_pass_prompt = ( "You are the Nubenetes Fast-Pass Evaluator (2026).\n" f"Analyze the following resource details:\n" f"- Title: {title}\n" f"- URL: {url}\n" f"- Context: {desc}\n" f"- Proposed Tags: {tags}\n\n" f"{system_mandates}\n\n" "Evaluate this resource across Security, SRE, and AI/Developer DX aspects and assign a score (0 to 100).\n" "Also generate a high-density, professional technical summary (2-5 sentences, O'Reilly technical style) and select appropriate tags.\n" "Respond ONLY in valid JSON format: {\"score\": int, \"justification\": \"string\", \"summary\": \"string\", \"tags\": [\"string\"]}" ) fast_pass_score = int(initial_score) fast_pass_justification = "Failed to run fast-pass." fast_pass_summary = desc fast_pass_tags = tags try: res = await call_gemini_with_retry(fast_pass_prompt, prefer_flash=True, use_grounding=True, role="FastPass-Evaluator") fast_pass_score = min(max(int(res.get("score", 50)), 0), 100) fast_pass_justification = res.get("justification", "No justification provided.") fast_pass_summary = res.get("summary", desc) fast_pass_tags = res.get("tags", tags) log_event(f" [🔍] Fast-Pass Evaluator rated score: {fast_pass_score}") except Exception as e: log_event(f" [!] Fast-Pass Evaluator failed: {e}") # Check if the score falls outside the borderline uncertainty margin [60, 75] if fast_pass_score >= 76 or fast_pass_score <= 59: log_event(f" [⚡] Fast-Pass Consensus reached ({fast_pass_score}). Skipping full debate panel!") debate_data = { "url": url, "title": title, "initial_score": initial_score, "final_consensus_score": fast_pass_score, "fast_pass": True, "justification": fast_pass_justification, "timestamp": datetime.now().isoformat(), "final_tags": fast_pass_tags, "refined_summary": fast_pass_summary, "content_hash": content_hash } try: memory_data = {} if os.path.exists(DEBATE_MEMORY_FILE): try: memory_data = json.load(open(DEBATE_MEMORY_FILE, "r")) except Exception as e: log_event(f"[WARN] load debate memory for fast-pass persist: {str(e)[:100]}") memory_data.setdefault("resolved_debates", {})[normalize_url(url)] = debate_data with open(DEBATE_MEMORY_FILE, "w") as f: json.dump(memory_data, f, indent=2) except Exception as e: log_event(f" [!] Failed to persist debate memory: {e}") return fast_pass_score, fast_pass_tags, fast_pass_summary, debate_data log_event(f" [⚖️] Borderline score detected ({fast_pass_score}). Escalating to full Multi-Agent Debate Panel...") # 1. Independent Evaluation Round personas = { "Security Architect": "Focus on licensing (MIT/Apache vs BSL/SSPL), supply chain security, access control, vulnerabilities, and enterprise compliance.", "Cloud Native SRE": "Focus on high-availability, scalability, production maturity, operational complexity, community health, and performance metrics.", "AI Platform Engineer": "Focus on developer agility, tooling simplicity, AI stack integration (MCP, LLMs, agents), and 2026 architectural relevance." } scores = {} justifications = {} async def evaluate_persona(name: str, focus: str) -> Tuple[int, str]: prompt = ( f"You are the Nubenetes {name}.\n" f"Your perspective: {focus}\n\n" f"{system_mandates}\n\n" f"Evaluate the following resource:\n" f"- Title: {title}\n" f"- URL: {url}\n" f"- Context: {desc}\n" f"- Proposed Tags: {tags}\n\n" "Assign an architectural impact score (0 to 100) and write a 1-2 sentence technical justification.\n" "Respond ONLY in valid JSON format: {\"score\": int, \"justification\": \"string\"}" ) try: res = await call_gemini_with_retry(prompt, prefer_flash=True, use_grounding=True, role=f"Debater-{name.replace(' ', '')}") score = min(max(int(res.get("score", 50)), 0), 100) justification = res.get("justification", "No justification provided.") return score, justification except Exception as e: log_event(f" [!] Persona {name} evaluation failed: {e}") return int(initial_score), "Failed to evaluate." # Run evaluations in parallel eval_tasks = [evaluate_persona(name, focus) for name, focus in personas.items()] eval_results = await asyncio.gather(*eval_tasks) for idx, (name, _) in enumerate(personas.items()): scores[name], justifications[name] = eval_results[idx] log_event(f" [>] {name} rated: {scores[name]} (Justification: {justifications[name]})") # Check divergence: if max diff >= 15 points, run a debate round max_score = max(scores.values()) min_score = min(scores.values()) divergence = max_score - min_score debate_transcript = [] if divergence >= 15: log_event(f" [⚖️] Divergence detected ({divergence} points). Starting Debate Round...") async def run_rebuttal(name: str, focus: str) -> Tuple[int, str]: opponent_views = "\n".join([f"- {other}: Score {scores[other]} | Justification: {justifications[other]}" for other in personas if other != name]) prompt = ( f"You are the Nubenetes {name}.\n" f"Your perspective: {focus}\n\n" "The panel of experts disagrees on this resource. Here are the other views:\n" f"{opponent_views}\n\n" f"Resource details:\n" f"- Title: {title}\n" f"- URL: {url}\n" f"- Context: {desc}\n\n" "Reconsider your evaluation. You can either defend your initial score or adjust it.\n" "Respond ONLY in valid JSON format: {\"score\": int, \"rebuttal\": \"1-2 sentence response to other experts\"}" ) try: res = await call_gemini_with_retry(prompt, prefer_flash=True, use_grounding=True, role=f"Debate-Rebuttal-{name.replace(' ', '')}") new_score = min(max(int(res.get("score", scores[name])), 0), 100) rebuttal = res.get("rebuttal", "No rebuttal provided.") return new_score, rebuttal except Exception as e: log_event(f" [!] Persona {name} rebuttal failed: {e}") return scores[name], "No rebuttal provided." rebuttal_tasks = [run_rebuttal(name, focus) for name, focus in personas.items()] rebuttal_results = await asyncio.gather(*rebuttal_tasks) for idx, (name, _) in enumerate(personas.items()): scores[name], rebuttal = rebuttal_results[idx] debate_transcript.append(f"{name} (Score {scores[name]}): {rebuttal}") log_event(f" [>] {name} revised rating to {scores[name]}. Rebuttal: {rebuttal}") # Compute Final Consensus final_score = int(sum(scores.values()) / len(scores)) log_event(f" [🏁] Consensus Score reached: {final_score}") # 2. Refined Curation/Tags Round refine_prompt = ( "You are the Nubenetes Curation Synthesis Agent (2026).\n" f"Combine the following multi-agent expert reviews into a final technical decision.\n\n" f"Resource Title: {title}\n" f"URL: {url}\n" f"Consensus Score: {final_score}\n" f"Security Architect Score: {scores['Security Architect']} | Justification: {justifications['Security Architect']}\n" f"Cloud Native SRE Score: {scores['Cloud Native SRE']} | Justification: {justifications['Cloud Native SRE']}\n" f"AI Platform Engineer Score: {scores['AI Platform Engineer']} | Justification: {justifications['AI Platform Engineer']}\n\n" "Generate a final, high-density, professional technical summary (2-5 sentences, HSL-themed style, no generic statements).\n" "Select the appropriate subset of tags. You MUST include:\n" "1. Standard maturity tags from: [DE FACTO STANDARD], [ENTERPRISE-STABLE], [EMERGING], [GUIDE], [CASE STUDY], [COMMUNITY-TOOL], [LEGACY].\n" "2. Any relevant fine-grained technology stack tags from the content (e.g., [EBPF], [WASM], [GITOPS], [IAC], [SERVICE-MESH], [SERVERLESS], [MLOPS], [DB]). Keep them uppercase and wrapped in brackets.\n" "Respond ONLY in valid JSON format: {\"summary\": \"refined summary...\", \"tags\": [\"...\"]}" ) refined_summary = desc final_tags = tags try: res = await call_gemini_with_retry(refine_prompt, prefer_flash=False, use_grounding=False, role="Debate-Synthesis") refined_summary = res.get("summary", desc) final_tags = res.get("tags", tags) except Exception as e: log_event(f" [!] Error during debate synthesis: {e}") debate_data = { "url": url, "title": title, "initial_score": initial_score, "final_consensus_score": final_score, "scores": scores, "justifications": justifications, "rebuttals": debate_transcript, "timestamp": datetime.now().isoformat(), "final_tags": final_tags, "refined_summary": refined_summary, "content_hash": content_hash } # Persist the resolved debate to memory log (Mandate 3.1) try: memory_data = {} if os.path.exists(DEBATE_MEMORY_FILE): try: memory_data = json.load(open(DEBATE_MEMORY_FILE, "r")) except Exception as e: log_event(f"[WARN] load debate memory for final persist: {str(e)[:100]}") memory_data.setdefault("resolved_debates", {})[normalize_url(url)] = debate_data # Keep blacklist and other fields intact with open(DEBATE_MEMORY_FILE, "w") as f: json.dump(memory_data, f, indent=2) except Exception as e: log_event(f" [!] Failed to persist debate memory: {e}") return final_score, final_tags, refined_summary, debate_data