""" Operon Convergence Compiler -- Compile & Verify ================================================ Build a multi-stage organism, compile it to an external framework (Swarms, DeerFlow, Ralph, Scion), and verify the structural certificates that the compiler preserves through the translation. Run locally: pip install gradio && python space-compiler/app.py """ import json import sys from pathlib import Path import gradio as gr _repo_root = Path(__file__).resolve().parents[2] if str(_repo_root) not in sys.path: sys.path.insert(0, str(_repo_root)) from operon_ai import ATP_Store, MockProvider, Nucleus, SkillStage, skill_organism from operon_ai.convergence import ( organism_to_deerflow, organism_to_ralph, organism_to_scion, organism_to_swarms, ) from operon_ai.core.certificate import verify_compiled # --------------------------------------------------------------------------- # Presets & constants # --------------------------------------------------------------------------- PRESETS: dict[str, dict] = { "(custom)": dict( s1_name="intake", s1_role="Normalizer", s2_name="router", s2_role="Classifier", s3_name="executor", s3_role="Analyst", framework="Swarms", budget=1000, zero_budget=False, ), "Research pipeline": dict( s1_name="gather", s1_role="Researcher", s2_name="synthesize", s2_role="Synthesizer", s3_name="report", s3_role="Writer", framework="DeerFlow", budget=1500, zero_budget=False, ), "Code review": dict( s1_name="diff_parser", s1_role="Parser", s2_name="reviewer", s2_role="Reviewer", s3_name="commenter", s3_role="Commentator", framework="Ralph", budget=800, zero_budget=False, ), "Failing certificate": dict( s1_name="intake", s1_role="Normalizer", s2_name="router", s2_role="Classifier", s3_name="executor", s3_role="Analyst", framework="Swarms", budget=0, zero_budget=True, ), } FRAMEWORKS = ["Swarms", "DeerFlow", "Ralph", "Scion"] _COMPILER_MAP = { "Swarms": organism_to_swarms, "DeerFlow": organism_to_deerflow, "Ralph": organism_to_ralph, "Scion": organism_to_scion, } # --------------------------------------------------------------------------- # Core logic # --------------------------------------------------------------------------- def _build_organism(names, roles, budget_val): """Build a 3-stage SkillOrganism with mock nuclei.""" fast = Nucleus(provider=MockProvider(responses={"classify": "ROUTE: analysis"})) deep = Nucleus(provider=MockProvider(responses={"analysis": "Analysis complete."})) return skill_organism( stages=[ SkillStage(name=names[0], role=roles[0], handler=lambda task, state, outputs, stage: {"request": task}), SkillStage(name=names[1], role=roles[1], instructions="Classify the incoming request.", mode="fixed"), SkillStage(name=names[2], role=roles[2], instructions="Analyze the request in depth.", mode="fuzzy"), ], fast_nucleus=fast, deep_nucleus=deep, budget=ATP_Store(budget=budget_val, silent=True), ) def _cert_badge(holds: bool) -> str: color, label = ("#22c55e", "HOLDS") if holds else ("#ef4444", "FAILS") return (f'{label}') def _build_cert_html(verifications): """Render certificate verification results as styled HTML.""" if not verifications: return ( '
' 'No certificates found in compiled output.
' '' 'Budget may be 0 -- the organism produces no metabolic certificates ' 'when there is no energy to gate.
Please fill in all stage names and roles.
" effective_budget = 0 if zero_budget else int(budget_val) try: org = _build_organism(names, roles, effective_budget) compiled = _COMPILER_MAP[framework](org) compiled_json = json.dumps(compiled, indent=2, default=str) verifications = verify_compiled(compiled) return compiled_json, _build_cert_html(verifications) except Exception as e: return json.dumps({"error": str(e)}, indent=2), ( f'Error
' f'{e}