app.py

"""
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'<span style="background:{color};color:white;padding:2px 8px;'
            f'border-radius:4px;font-size:0.85em;font-weight:600;">{label}</span>')


def _build_cert_html(verifications):
    """Render certificate verification results as styled HTML."""
    if not verifications:
        return (
            '<div style="padding:16px;border-radius:8px;border:2px solid #f59e0b;'
            'background:#fffbeb;">'
            '<p style="margin:0;font-weight:600;color:#92400e;">'
            'No certificates found in compiled output.</p>'
            '<p style="margin:4px 0 0;color:#78716c;font-size:0.9em;">'
            'Budget may be 0 -- the organism produces no metabolic certificates '
            'when there is no energy to gate.</p></div>')

    all_hold = all(v.holds for v in verifications)
    bc = "#22c55e" if all_hold else "#ef4444"
    bg = "#f0fdf4" if all_hold else "#fef2f2"
    summary = "All certificates verified" if all_hold else "One or more certificates failed"
    n = len(verifications)

    rows = ""
    for v in verifications:
        ev = ", ".join(f"<b>{k}</b>={val}" for k, val in dict(v.evidence).items())
        rows += (
            f'<div style="padding:8px 12px;border-bottom:1px solid #e5e7eb;">'
            f'<div style="display:flex;align-items:center;gap:10px;">'
            f'{_cert_badge(v.holds)}'
            f'<span style="font-weight:600;">{v.certificate.theorem}</span></div>'
            f'<div style="margin-top:4px;font-size:0.85em;color:#6b7280;">'
            f'{v.certificate.conclusion}</div>'
            f'<div style="margin-top:4px;font-size:0.85em;color:#374151;">'
            f'Evidence: {ev}</div></div>')

    return (
        f'<div style="padding:0;border-radius:8px;border:2px solid {bc};'
        f'background:{bg};overflow:hidden;">'
        f'<div style="padding:12px 16px;background:{bc}10;border-bottom:1px solid {bc};">'
        f'<span style="font-weight:700;font-size:1.05em;">{summary}</span>'
        f'<span style="margin-left:8px;color:#6b7280;font-size:0.9em;">'
        f'({n} certificate{"s" if n != 1 else ""})</span></div>'
        f'{rows}</div>')


def compile_and_verify(
    s1_name, s1_role, s2_name, s2_role, s3_name, s3_role,
    framework, budget_val, zero_budget,
):
    """Compile the organism and verify certificates."""
    names = [s.strip() for s in (s1_name, s2_name, s3_name)]
    roles = [s.strip() for s in (s1_role, s2_role, s3_role)]
    if not all(names) or not all(roles):
        return "{}", "<p>Please fill in all stage names and roles.</p>"

    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'<div style="padding:16px;border-radius:8px;border:2px solid #ef4444;'
            f'background:#fef2f2;">'
            f'<p style="margin:0;font-weight:600;color:#991b1b;">Error</p>'
            f'<p style="margin:4px 0 0;color:#7f1d1d;">{e}</p></div>')


def load_preset(name: str):
    p = PRESETS.get(name)
    if not p:
        return "intake", "Normalizer", "router", "Classifier", "executor", "Analyst", "Swarms", 1000, False
    return (p["s1_name"], p["s1_role"], p["s2_name"], p["s2_role"],
            p["s3_name"], p["s3_role"], p["framework"], p["budget"], p["zero_budget"])

# ---------------------------------------------------------------------------
# Gradio UI
# ---------------------------------------------------------------------------

def build_app() -> gr.Blocks:
    with gr.Blocks(title="Operon Convergence Compiler") as app:
        gr.Markdown(
            "# Operon Convergence Compiler\n"
            "Build a 3-stage organism, compile it to an external agent framework, "
            "and verify that **structural certificates** (budget gating, priority "
            "guarantees) survive the translation.\n\n"
            "[GitHub](https://github.com/coredipper/operon) | "
            "[Paper](https://github.com/coredipper/operon/tree/main/article)")

        with gr.Row():
            preset_dropdown = gr.Dropdown(
                choices=list(PRESETS.keys()), value="(custom)",
                label="Load Preset", scale=2)
            compile_btn = gr.Button("Compile", variant="primary", scale=1)

        gr.Markdown("### Pipeline Stages")
        with gr.Row():
            s1_name = gr.Textbox(value="intake", label="Stage 1 Name", scale=1)
            s1_role = gr.Textbox(value="Normalizer", label="Stage 1 Role", scale=1)
        with gr.Row():
            s2_name = gr.Textbox(value="router", label="Stage 2 Name", scale=1)
            s2_role = gr.Textbox(value="Classifier", label="Stage 2 Role", scale=1)
        with gr.Row():
            s3_name = gr.Textbox(value="executor", label="Stage 3 Name", scale=1)
            s3_role = gr.Textbox(value="Analyst", label="Stage 3 Role", scale=1)

        with gr.Row():
            framework_dd = gr.Dropdown(
                choices=FRAMEWORKS, value="Swarms",
                label="Target Framework", scale=1)
            budget_slider = gr.Slider(
                minimum=0, maximum=2000, value=1000, step=50,
                label="ATP Budget", scale=1)
            zero_budget_cb = gr.Checkbox(
                value=False, label="Set budget to 0 (show failing certificate)")

        gr.Markdown("### Output")
        with gr.Row():
            with gr.Column(scale=1):
                gr.Markdown("#### Compiled JSON")
                compiled_output = gr.Code(language="json", label="Compiled output")
            with gr.Column(scale=1):
                gr.Markdown("#### Certificate Verification")
                cert_output = gr.HTML()

        all_inputs = [s1_name, s1_role, s2_name, s2_role, s3_name, s3_role,
                      framework_dd, budget_slider, zero_budget_cb]

        compile_btn.click(fn=compile_and_verify, inputs=all_inputs,
                          outputs=[compiled_output, cert_output])
        preset_dropdown.change(fn=load_preset, inputs=[preset_dropdown],
                               outputs=all_inputs)
    return app


if __name__ == "__main__":
    app = build_app()
    app.launch(theme=gr.themes.Soft())