app.py

"""
Operon Signal Cascade -- Multi-Stage Amplification Pipeline
===========================================================

Feed text through a multi-stage processing cascade with amplification
factors and checkpoint gates, or explore the classic MAPK three-tier
biological signaling pattern.

Run locally:
    pip install gradio
    python space-cascade/app.py
"""

import sys
import time
from pathlib import Path

import gradio as gr

_repo_root = Path(__file__).resolve().parent.parent
if str(_repo_root) not in sys.path:
    sys.path.insert(0, str(_repo_root))

from operon_ai import (
    Cascade,
    CascadeStage,
    CascadeResult,
    StageResult,
    StageStatus,
    MAPKCascade,
)


# ---------------------------------------------------------------------------
# Presets
# ---------------------------------------------------------------------------

PRESETS: dict[str, dict] = {
    "(custom)": {"text": "", "description": "Type your own input"},
    "Safe text": {
        "text": "Hello World From Operon",
        "description": "Clean text passes all stages",
    },
    "XSS attempt": {
        "text": "Hello <script>alert('xss')</script>",
        "description": "Blocked at sanitize checkpoint -- HTML tags stripped",
    },
    "Empty input": {
        "text": "",
        "description": "Blocked at validate stage -- empty/None input rejected",
    },
    "Long text": {
        "text": (
            "The Operon framework models AI agent infrastructure using biological "
            "metaphors. Each organelle serves a distinct function: the Membrane "
            "filters threats, the Mitochondria powers computation, the Ribosome "
            "synthesizes prompts, the Chaperone validates output structure, and "
            "the Lysosome recycles waste from failed operations."
        ),
        "description": "Paragraph of text -- full pipeline with high token count",
    },
    "Single word": {
        "text": "hello",
        "description": "Minimal input -- low amplification, fast pipeline",
    },
    "Unicode text": {
        "text": "Bonjour le monde! \u2603 \u2764",
        "description": "Unicode characters pass normalization cleanly",
    },
    "Special characters": {
        "text": "test@email.com & 100% success <guaranteed>",
        "description": "Mixed special chars -- sanitize strips angle brackets",
    },
    "MAPK preset": {
        "text": "signal",
        "description": "Run with the MAPK tab for biological cascade demo",
    },
}


# ---------------------------------------------------------------------------
# Styling
# ---------------------------------------------------------------------------

STATUS_COLORS = {
    StageStatus.COMPLETED: ("#22c55e", "COMPLETED"),
    StageStatus.FAILED: ("#ef4444", "FAILED"),
    StageStatus.BLOCKED: ("#f59e0b", "BLOCKED"),
    StageStatus.SKIPPED: ("#6b7280", "SKIPPED"),
    StageStatus.PENDING: ("#94a3b8", "PENDING"),
    StageStatus.RUNNING: ("#3b82f6", "RUNNING"),
}


def _status_badge(status: StageStatus) -> str:
    color, label = STATUS_COLORS.get(status, ("#6b7280", "UNKNOWN"))
    return (
        f'<span style="background:{color};color:white;padding:2px 8px;'
        f'border-radius:4px;font-size:0.85em;font-weight:600;">{label}</span>'
    )


# ---------------------------------------------------------------------------
# Text pipeline processors
# ---------------------------------------------------------------------------

def _validate(text):
    """Validate input is non-empty."""
    if text is None or (isinstance(text, str) and not text.strip()):
        raise ValueError("Empty input")
    return text


def _normalize(text):
    """Normalize whitespace and case."""
    return " ".join(str(text).strip().split())


def _tokenize(text):
    """Split into tokens."""
    tokens = str(text).split()
    return {"text": text, "tokens": tokens, "count": len(tokens)}


def _sanitize_check(text) -> bool:
    """Checkpoint: reject input with HTML/script tags."""
    s = str(text)
    return "<script" not in s.lower() and "</script" not in s.lower()


def _filter_output(data):
    """Final filter -- clean output."""
    if isinstance(data, dict):
        text = data.get("text", "")
        # Strip any remaining HTML-like tags
        import re
        clean = re.sub(r'<[^>]+>', '', str(text))
        data["text"] = clean
        data["clean"] = True
        return data
    return {"text": str(data), "tokens": str(data).split(), "count": len(str(data).split()), "clean": True}


def _build_text_cascade() -> Cascade:
    """Build the text processing cascade."""
    cascade = Cascade(name="TextPipeline", halt_on_failure=True, silent=True)
    cascade.add_stage(CascadeStage(
        name="validate",
        processor=_validate,
        amplification=1.0,
        checkpoint=lambda x: x is not None and (not isinstance(x, str) or len(x.strip()) > 0),
    ))
    cascade.add_stage(CascadeStage(
        name="normalize",
        processor=_normalize,
        amplification=1.0,
    ))
    cascade.add_stage(CascadeStage(
        name="sanitize",
        processor=lambda x: x,  # pass-through; checkpoint does the work
        amplification=1.5,
        checkpoint=_sanitize_check,
    ))
    cascade.add_stage(CascadeStage(
        name="tokenize",
        processor=_tokenize,
        amplification=2.0,
    ))
    cascade.add_stage(CascadeStage(
        name="filter",
        processor=_filter_output,
        amplification=1.0,
    ))
    return cascade


# ---------------------------------------------------------------------------
# Core logic -- Text Pipeline
# ---------------------------------------------------------------------------

def run_text_pipeline(preset_name: str, custom_text: str) -> tuple[str, str, str]:
    """Run the text cascade pipeline.

    Returns (result_html, trace_html, summary_md).
    """
    text = custom_text.strip() if custom_text.strip() else PRESETS.get(preset_name, {}).get("text", "")

    cascade = _build_text_cascade()
    result: CascadeResult = cascade.run(text if text else None)

    # --- Result banner ---
    if result.success:
        final = result.final_output
        if isinstance(final, dict):
            display = final.get("text", str(final))
            token_count = final.get("count", "?")
        else:
            display = str(final)
            token_count = len(str(final).split())

        result_html = (
            f'<div style="padding:16px;border-radius:8px;border:2px solid #22c55e;background:#f0fdf4;">'
            f'<div style="font-size:1.3em;font-weight:700;color:#16a34a;margin-bottom:8px;">'
            f'Cascade Complete</div>'
            f'<div style="font-family:monospace;color:#15803d;font-size:1.1em;margin:8px 0;">'
            f'{display}</div>'
            f'<div style="font-size:0.85em;color:#6b7280;">'
            f'Tokens: {token_count} | '
            f'Total amplification: {result.total_amplification:.1f}x | '
            f'Time: {result.total_time_ms:.2f} ms</div>'
            f'</div>'
        )
    else:
        blocked_at = result.blocked_at or "Unknown"
        result_html = (
            f'<div style="padding:16px;border-radius:8px;border:2px solid #ef4444;background:#fef2f2;">'
            f'<div style="font-size:1.3em;font-weight:700;color:#dc2626;margin-bottom:8px;">'
            f'Cascade Blocked at: {blocked_at}</div>'
            f'<div style="font-size:0.9em;color:#991b1b;">'
            f'Completed {result.stages_completed}/{result.stages_total} stages | '
            f'Time: {result.total_time_ms:.2f} ms</div>'
            f'</div>'
        )

    # --- Stage trace ---
    trace_html = ""
    stage_colors = ["#3b82f6", "#8b5cf6", "#f59e0b", "#22c55e", "#6366f1"]
    for i, sr in enumerate(result.stage_results):
        color = stage_colors[i % len(stage_colors)]
        status_color = "#16a34a" if sr.status == StageStatus.COMPLETED else "#dc2626"

        input_preview = str(sr.input_signal)[:80] + ("..." if len(str(sr.input_signal)) > 80 else "")
        output_preview = str(sr.output_signal)[:80] + ("..." if len(str(sr.output_signal)) > 80 else "")

        trace_html += (
            f'<div style="border-left:4px solid {color};padding:8px 12px;margin:8px 0;'
            f'background:#f9fafb;border-radius:0 4px 4px 0;">'
            f'<div style="display:flex;justify-content:space-between;align-items:center;">'
            f'<span style="font-weight:700;color:{color};">'
            f'Stage {i+1}: {sr.stage_name}</span>'
            f'{_status_badge(sr.status)}'
            f'</div>'
            f'<div style="font-size:0.85em;color:#374151;margin-top:4px;">'
            f'In: <code>{input_preview}</code></div>'
            f'<div style="font-size:0.85em;color:#374151;">'
            f'Out: <code>{output_preview}</code></div>'
            f'<div style="font-size:0.8em;color:#6b7280;margin-top:2px;">'
            f'Amplification: {sr.amplification_factor:.1f}x | '
            f'Time: {sr.processing_time_ms:.3f} ms'
        )
        if sr.error:
            trace_html += f' | Error: <span style="color:#dc2626;">{sr.error}</span>'
        trace_html += '</div></div>'

    # --- Summary ---
    summary_md = "### Pipeline Summary\n\n"
    summary_md += f"**Stages completed:** {result.stages_completed}/{result.stages_total}\n\n"
    summary_md += f"**Total amplification:** {result.total_amplification:.1f}x\n\n"
    summary_md += f"**Total time:** {result.total_time_ms:.2f} ms\n\n"
    if result.blocked_at:
        summary_md += f"**Blocked at:** {result.blocked_at}\n\n"

    summary_md += "### Stage Roles\n\n"
    summary_md += "| Stage | Role |\n"
    summary_md += "|-------|------|\n"
    summary_md += "| validate | Reject empty/None input |\n"
    summary_md += "| normalize | Normalize whitespace |\n"
    summary_md += "| sanitize | Checkpoint blocks script tags |\n"
    summary_md += "| tokenize | Split into tokens (2x amplification) |\n"
    summary_md += "| filter | Final cleanup pass |\n"

    return result_html, trace_html, summary_md


# ---------------------------------------------------------------------------
# Core logic -- MAPK Cascade
# ---------------------------------------------------------------------------

def run_mapk(tier1_amp: float, tier2_amp: float, tier3_amp: float) -> tuple[str, str, str]:
    """Run the MAPK cascade with configurable amplification.

    Returns (result_html, trace_html, summary_md).
    """
    mapk = MAPKCascade(
        name="MAPK-Demo",
        tier1_amplification=tier1_amp,
        tier2_amplification=tier2_amp,
        tier3_amplification=tier3_amp,
        silent=True,
    )

    result: CascadeResult = mapk.run("signal")
    total_amp = tier1_amp * tier2_amp * tier3_amp

    # --- Result banner ---
    result_html = (
        f'<div style="padding:16px;border-radius:8px;border:2px solid #8b5cf6;background:#f5f3ff;">'
        f'<div style="font-size:1.3em;font-weight:700;color:#7c3aed;margin-bottom:8px;">'
        f'MAPK Cascade Result</div>'
        f'<div style="font-family:monospace;color:#5b21b6;font-size:1.1em;margin:8px 0;">'
        f'Final output: {result.final_output}</div>'
        f'<div style="display:flex;gap:20px;font-size:0.9em;color:#6b7280;flex-wrap:wrap;">'
        f'<span>Configured amplification: <b>{total_amp:.0f}x</b></span>'
        f'<span>Actual amplification: <b>{result.total_amplification:.1f}x</b></span>'
        f'<span>Time: <b>{result.total_time_ms:.2f} ms</b></span>'
        f'</div>'
        f'</div>'
    )

    # --- Stage trace ---
    tier_names = ["MAPKKK (Tier 1)", "MAPKK (Tier 2)", "MAPK (Tier 3)"]
    tier_colors = ["#ef4444", "#f59e0b", "#22c55e"]
    trace_html = ""
    for i, sr in enumerate(result.stage_results):
        color = tier_colors[i] if i < len(tier_colors) else "#6b7280"
        label = tier_names[i] if i < len(tier_names) else sr.stage_name

        trace_html += (
            f'<div style="border-left:4px solid {color};padding:8px 12px;margin:8px 0;'
            f'background:#f9fafb;border-radius:0 4px 4px 0;">'
            f'<div style="display:flex;justify-content:space-between;align-items:center;">'
            f'<span style="font-weight:700;color:{color};">{label}</span>'
            f'{_status_badge(sr.status)}'
            f'</div>'
            f'<div style="font-size:0.85em;color:#374151;margin-top:4px;">'
            f'In: <code>{sr.input_signal}</code> | Out: <code>{sr.output_signal}</code></div>'
            f'<div style="font-size:0.8em;color:#6b7280;margin-top:2px;">'
            f'Amplification: {sr.amplification_factor:.1f}x | '
            f'Time: {sr.processing_time_ms:.3f} ms</div>'
            f'</div>'
        )

    # --- Summary ---
    summary_md = "### MAPK Signaling\n\n"
    summary_md += f"**Tier 1 (MAPKKK):** {tier1_amp:.0f}x amplification\n\n"
    summary_md += f"**Tier 2 (MAPKK):** {tier2_amp:.0f}x amplification\n\n"
    summary_md += f"**Tier 3 (MAPK):** {tier3_amp:.0f}x amplification\n\n"
    summary_md += f"**Total:** {tier1_amp:.0f} x {tier2_amp:.0f} x {tier3_amp:.0f} = **{total_amp:.0f}x**\n\n"
    summary_md += "### How MAPK Works\n\n"
    summary_md += "The MAPK (Mitogen-Activated Protein Kinase) cascade is a "
    summary_md += "three-tier signaling pathway found in all eukaryotic cells. "
    summary_md += "Each tier amplifies the signal from the previous tier, "
    summary_md += "enabling a small extracellular signal to produce a large "
    summary_md += "intracellular response.\n"

    return result_html, trace_html, summary_md


def load_preset(name: str) -> str:
    return PRESETS.get(name, {}).get("text", "")


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

def build_app() -> gr.Blocks:
    with gr.Blocks(title="Operon Signal Cascade") as app:
        gr.Markdown(
            "# Operon Signal Cascade\n"
            "Multi-stage signal processing with **amplification factors** and "
            "**checkpoint gates**. Each stage transforms the signal; checkpoints "
            "can block progression.\n\n"
            "[GitHub](https://github.com/coredipper/operon) | "
            "[Paper](https://github.com/coredipper/operon/tree/main/article)"
        )

        with gr.Tabs():
            with gr.TabItem("Text Pipeline"):
                with gr.Row():
                    preset_dropdown = gr.Dropdown(
                        choices=list(PRESETS.keys()),
                        value="(custom)",
                        label="Load Preset",
                        scale=2,
                    )
                    run_btn = gr.Button("Run Cascade", variant="primary", scale=1)

                text_input = gr.Textbox(
                    label="Input Text",
                    placeholder="Type text to process through the cascade...",
                    lines=3,
                )

                result_html = gr.HTML(label="Result")

                with gr.Row():
                    with gr.Column(scale=2):
                        gr.Markdown("### Stage Trace")
                        trace_html = gr.HTML()
                    with gr.Column(scale=1):
                        summary_md = gr.Markdown()

                run_btn.click(
                    fn=run_text_pipeline,
                    inputs=[preset_dropdown, text_input],
                    outputs=[result_html, trace_html, summary_md],
                )
                text_input.submit(
                    fn=run_text_pipeline,
                    inputs=[preset_dropdown, text_input],
                    outputs=[result_html, trace_html, summary_md],
                )
                preset_dropdown.change(
                    fn=load_preset,
                    inputs=[preset_dropdown],
                    outputs=[text_input],
                )

            with gr.TabItem("MAPK Cascade"):
                gr.Markdown(
                    "### MAPK Three-Tier Signaling\n\n"
                    "The classic MAPKKK > MAPKK > MAPK biological cascade. "
                    "Configure amplification at each tier and see how signal "
                    "strength compounds through the pathway."
                )

                with gr.Row():
                    tier1_slider = gr.Slider(
                        minimum=1, maximum=20, value=10, step=1,
                        label="Tier 1 (MAPKKK) Amplification",
                    )
                    tier2_slider = gr.Slider(
                        minimum=1, maximum=20, value=10, step=1,
                        label="Tier 2 (MAPKK) Amplification",
                    )
                    tier3_slider = gr.Slider(
                        minimum=1, maximum=20, value=10, step=1,
                        label="Tier 3 (MAPK) Amplification",
                    )

                mapk_btn = gr.Button("Run MAPK Cascade", variant="primary")

                mapk_result_html = gr.HTML(label="MAPK Result")

                with gr.Row():
                    with gr.Column(scale=2):
                        gr.Markdown("### Tier Trace")
                        mapk_trace_html = gr.HTML()
                    with gr.Column(scale=1):
                        mapk_summary_md = gr.Markdown()

                mapk_btn.click(
                    fn=run_mapk,
                    inputs=[tier1_slider, tier2_slider, tier3_slider],
                    outputs=[mapk_result_html, mapk_trace_html, mapk_summary_md],
                )

    return app


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