app.py

"""Methodology Functor — live demo for Hugging Face Spaces.

Self-contained Gradio app. Bundles the minimal categorical types from
analysis/__init__.py and the methodology-DAG primitives from
analysis/methodology_dag.py so the Space deploys as a single file.

Source-of-truth lives in zero-rl-pipeline/analysis/methodology_dag.py.

Tej Desai x Claude Opus 4.7 (1M) — May 2026
Intuition Labs LLC
"""
from __future__ import annotations

from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable

import gradio as gr


# =========================================================================
# Bundled categorical types (mirror analysis/__init__.py)
# =========================================================================

KAPPA_SIGMA: float = 4.0


@dataclass(frozen=True)
class Object:
    name: str
    domain: str
    data: Any = None


@dataclass(frozen=True)
class Morphism:
    name: str
    source: Object
    target: Object
    domain: str
    transform: Callable | None = None


@dataclass
class Category:
    name: str
    objects: list[Object] = field(default_factory=list)
    morphisms: list[Morphism] = field(default_factory=list)
    identities: dict[str, Morphism] = field(default_factory=dict)

    def add_object(self, name: str, data: Any = None) -> Object:
        obj = Object(name=name, domain=self.name, data=data)
        self.objects.append(obj)
        self.identities[name] = Morphism(
            name=f"id_{name}", source=obj, target=obj,
            domain=self.name, transform=lambda x: x,
        )
        return obj

    def add_morphism(self, name: str, source: Object, target: Object) -> Morphism:
        m = Morphism(name=name, source=source, target=target, domain=self.name)
        self.morphisms.append(m)
        return m


@dataclass
class Functor:
    name: str
    source_cat: Category
    target_cat: Category
    object_map: dict[str, str] = field(default_factory=dict)
    morphism_map: dict[str, str] = field(default_factory=dict)

    def map_object(self, obj: Object) -> Object | None:
        tgt_name = self.object_map.get(obj.name)
        if tgt_name is None:
            return None
        for o in self.target_cat.objects:
            if o.name == tgt_name:
                return o
        return None

    def map_morphism(self, m: Morphism) -> Morphism | None:
        tgt_name = self.morphism_map.get(m.name)
        if tgt_name is None:
            return None
        for mm in self.target_cat.morphisms:
            if mm.name == tgt_name:
                return mm
        return None

    def verify_identity_preservation(self) -> list[tuple[str, bool]]:
        results = []
        for obj in self.source_cat.objects:
            src_id = self.source_cat.identities.get(obj.name)
            if src_id is None:
                continue
            mapped_obj = self.map_object(obj)
            if mapped_obj is None:
                results.append((obj.name, False))
                continue
            tgt_id = self.target_cat.identities.get(mapped_obj.name)
            mapped_id = self.map_morphism(src_id)
            ok = (mapped_id is not None and tgt_id is not None and
                  mapped_id.source == tgt_id.source and
                  mapped_id.target == tgt_id.target)
            results.append((obj.name, ok))
        return results


# =========================================================================
# Methodology layer
# =========================================================================

class MethodologyNode(str, Enum):
    PROBLEM = "problem"
    SUBSTRATE = "substrate"
    COUPLING = "coupling"
    CONTROL = "control"
    MEASUREMENT = "measurement"
    CHARACTERIZATION = "characterization"
    LIFT = "lift"
    WITNESS = "witness"


@dataclass(frozen=True)
class MethodologyAnnotation:
    node_type: MethodologyNode
    description: str = ""
    artifacts: tuple[str, ...] = ()


def methodology_object(cat: Category, name: str, node_type: MethodologyNode,
                       description: str = "", artifacts: tuple[str, ...] = ()) -> Object:
    return cat.add_object(name, data=MethodologyAnnotation(node_type, description, artifacts))


def node_type_of(obj: Object) -> MethodologyNode | None:
    if isinstance(obj.data, MethodologyAnnotation):
        return obj.data.node_type
    return None


def _register_identities(cat: Category) -> None:
    existing = {m.name for m in cat.morphisms}
    for id_morph in cat.identities.values():
        if id_morph.name not in existing:
            cat.morphisms.append(id_morph)


def _add_identity_maps(F: Functor) -> None:
    for src_name, tgt_name in F.object_map.items():
        F.morphism_map.setdefault(f"id_{src_name}", f"id_{tgt_name}")


# =========================================================================
# Encoded DAGs — Vortex (Nature 2026), Kuramoto (1975), CARL (Desai 2026)
# =========================================================================

def build_vortex() -> Category:
    cat = Category(name="Vortex")
    p = methodology_object(cat, "vortex_unreadable", MethodologyNode.PROBLEM,
        "Vortex states in superconductors lack coherent manipulation/readout.")
    s = methodology_object(cat, "granular_Al", MethodologyNode.SUBSTRATE,
        "Granular aluminium film provides pinning landscape for vortices.",
        ("granular Al film", "disorder pinning sites"))
    c = methodology_object(cat, "transmon_ancilla", MethodologyNode.COUPLING,
        "Transmon coupled to vortex flux for non-destructive readout.",
        ("transmon qubit", "flux-charge coupling"))
    co = methodology_object(cat, "microwave_drive", MethodologyNode.CONTROL,
        "Microwave drive applies Rabi/Ramsey rotations.",
        ("microwave pulse sequencer",))
    m = methodology_object(cat, "dispersive_readout", MethodologyNode.MEASUREMENT,
        "Dispersive readout of transmon yields vortex state non-destructively.",
        ("dispersive readout chain",))
    ch = methodology_object(cat, "T1_T2_times", MethodologyNode.CHARACTERIZATION,
        "Coherence times T1 (relaxation) and T2 (dephasing) measured.")
    w = methodology_object(cat, "us_coherence", MethodologyNode.WITNESS,
        "Microsecond-range coherent control demonstrated.")
    for name, src, tgt in [("frame", p, s), ("attach", s, c), ("drive", c, co),
                           ("probe", co, m), ("characterize", m, ch), ("certify", ch, w)]:
        cat.add_morphism(name, src, tgt)
    _register_identities(cat)
    return cat


def build_kuramoto() -> Category:
    cat = Category(name="Kuramoto")
    p = methodology_object(cat, "oscillators_desync", MethodologyNode.PROBLEM,
        "Population of coupled oscillators desynchronizes without coupling.")
    s = methodology_object(cat, "network_topology", MethodologyNode.SUBSTRATE,
        "Network topology provides coupling landscape.",
        ("adjacency matrix", "degree distribution"))
    c = methodology_object(cat, "kuramoto_K", MethodologyNode.COUPLING,
        "Kuramoto coupling K injects phase-attraction between oscillator pairs.",
        ("K coupling constant", "sin(theta_j - theta_i)"))
    co = methodology_object(cat, "frequency_dispersion", MethodologyNode.CONTROL,
        "Frequency dispersion g(omega) controls partial vs full sync regimes.",
        ("Lorentzian g(omega)", "K/Delta scan"))
    m = methodology_object(cat, "order_parameter_R", MethodologyNode.MEASUREMENT,
        "Complex order parameter R*e^(i*psi) = (1/N) sum exp(i*theta_j).",
        ("|R|", "psi"))
    ch = methodology_object(cat, "K_critical", MethodologyNode.CHARACTERIZATION,
        "Critical coupling K_c = 2/(pi*g(0)) marks synchronization onset.")
    w = methodology_object(cat, "synchrony_onset", MethodologyNode.WITNESS,
        "R > 0.5 demonstrates partial synchrony.")
    for name, src, tgt in [("frame", p, s), ("attach", s, c), ("drive", c, co),
                           ("probe", co, m), ("characterize", m, ch), ("certify", ch, w)]:
        cat.add_morphism(name, src, tgt)
    _register_identities(cat)
    return cat


def build_carl() -> Category:
    cat = Category(name="CARL")
    p = methodology_object(cat, "semantic_drift", MethodologyNode.PROBLEM,
        "Semantic frames decohere across long contexts; tools/format collapse.")
    s = methodology_object(cat, "residual_token_grid", MethodologyNode.SUBSTRATE,
        "Residual stream + token grid + anchors as pinning landscape.",
        ("residual stream", "token vocabulary", "CLAUDE.md anchors"))
    c = methodology_object(cat, "sandbox_ancilla", MethodologyNode.COUPLING,
        "Tool calls (CodingSandboxEnv) couple frame to external probe substrate.",
        ("CodingSandboxEnv", "tool-call ancilla protocol"))
    co = methodology_object(cat, "grpo_cascade_gate", MethodologyNode.CONTROL,
        "GRPO updates + cascade gate apply discipline-locked rotations.",
        ("GRPOConfig", "cascade gate", "tau-LR"))
    m = methodology_object(cat, "coherence_trace", MethodologyNode.MEASUREMENT,
        "Token-by-token CoherenceTrace + sematon trace as weak measurement.",
        ("CoherenceTrace", "sematon trace", "Trackio"))
    ch = methodology_object(cat, "tau_lr_kappa", MethodologyNode.CHARACTERIZATION,
        "tau-LR phase characterization, frac_reward_zero_std, length trajectory.")
    w = methodology_object(cat, "gated_carl_closure", MethodologyNode.WITNESS,
        "gated_carl ramps after gate; kappa*sigma=4 closure observed in trace.")
    for name, src, tgt in [("frame", p, s), ("attach", s, c), ("drive", c, co),
                           ("probe", co, m), ("characterize", m, ch), ("certify", ch, w)]:
        cat.add_morphism(name, src, tgt)
    _register_identities(cat)
    return cat


def build_functor(src: Category, tgt: Category, name: str,
                  obj_map: dict[str, str]) -> Functor:
    F = Functor(name=name, source_cat=src, target_cat=tgt, object_map=obj_map,
                morphism_map={"frame": "frame", "attach": "attach", "drive": "drive",
                              "probe": "probe", "characterize": "characterize",
                              "certify": "certify"})
    _add_identity_maps(F)
    return F


VORTEX_TO_CARL_MAP = {
    "vortex_unreadable": "semantic_drift",
    "granular_Al": "residual_token_grid",
    "transmon_ancilla": "sandbox_ancilla",
    "microwave_drive": "grpo_cascade_gate",
    "dispersive_readout": "coherence_trace",
    "T1_T2_times": "tau_lr_kappa",
    "us_coherence": "gated_carl_closure",
}

KURAMOTO_TO_CARL_MAP = {
    "oscillators_desync": "semantic_drift",
    "network_topology": "residual_token_grid",
    "kuramoto_K": "sandbox_ancilla",
    "frequency_dispersion": "grpo_cascade_gate",
    "order_parameter_R": "coherence_trace",
    "K_critical": "tau_lr_kappa",
    "synchrony_onset": "gated_carl_closure",
}


# =========================================================================
# Verification
# =========================================================================

@dataclass
class FunctorWitness:
    object_coverage: float
    morphism_coverage: float
    type_preservation: float
    identity_preservation: float
    closure: float


def verify_methodology_functor(F: Functor) -> FunctorWitness:
    src = F.source_cat
    obj_total = len(src.objects)
    obj_mapped = sum(1 for o in src.objects if F.map_object(o) is not None)
    object_coverage = obj_mapped / max(obj_total, 1)

    real_morphs = [m for m in src.morphisms if not m.name.startswith("id_")]
    morph_mapped = sum(1 for m in real_morphs if F.map_morphism(m) is not None)
    morphism_coverage = morph_mapped / max(len(real_morphs), 1)

    type_ok, type_checked = 0, 0
    for m in real_morphs:
        fm = F.map_morphism(m)
        if fm is None:
            continue
        type_checked += 1
        if (node_type_of(m.source) == node_type_of(fm.source) and
                node_type_of(m.target) == node_type_of(fm.target)):
            type_ok += 1
    type_preservation = type_ok / max(type_checked, 1)

    id_results = F.verify_identity_preservation()
    id_ok = sum(1 for _, ok in id_results if ok)
    identity_preservation = id_ok / max(len(id_results), 1)

    closure = min(object_coverage, morphism_coverage,
                  type_preservation, identity_preservation)
    return FunctorWitness(object_coverage, morphism_coverage,
                          type_preservation, identity_preservation, closure)


# =========================================================================
# Registry
# =========================================================================

_VORTEX = build_vortex()
_KURAMOTO = build_kuramoto()
_CARL = build_carl()
_REGISTRY: dict[str, tuple[Category, Functor]] = {
    "Vortex (Nature 2026)": (_VORTEX, build_functor(_VORTEX, _CARL,
        "vortex_to_carl", VORTEX_TO_CARL_MAP)),
    "Kuramoto (1975)": (_KURAMOTO, build_functor(_KURAMOTO, _CARL,
        "kuramoto_to_carl", KURAMOTO_TO_CARL_MAP)),
}


# =========================================================================
# Gradio UI
# =========================================================================

PAYMENT_URL = "https://buy.stripe.com/4gM14n2Pm8jJcLx55U5EY00"


def render_dag_md(cat: Category) -> str:
    lines = [f"### {cat.name}", "", "| node type | name | description |", "|---|---|---|"]
    for obj in cat.objects:
        nt = node_type_of(obj)
        annot = obj.data
        type_str = nt.value if nt else "?"
        desc = annot.description if isinstance(annot, MethodologyAnnotation) else ""
        lines.append(f"| `{type_str}` | `{obj.name}` | {desc} |")
    return "\n".join(lines)


def render_functor_md(F: Functor) -> str:
    lines = [f"### Functor `{F.name}`", "", "| source object | target object |", "|---|---|"]
    src_order = [o.name for o in F.source_cat.objects]
    for src in src_order:
        tgt = F.object_map.get(src, "—")
        lines.append(f"| `{src}` | `{tgt}` |")
    return "\n".join(lines)


def render_witness_md(w: FunctorWitness) -> str:
    status = "PASS" if w.closure == 1.0 else ("PARTIAL" if w.closure > 0 else "FAIL")
    return f"""### Verification witness — **{status}**

| metric | value |
|---|---|
| object coverage | {w.object_coverage:.3f} |
| morphism coverage | {w.morphism_coverage:.3f} |
| type preservation | {w.type_preservation:.3f} |
| identity preservation | {w.identity_preservation:.3f} |
| **closure** | **{w.closure:.3f}** |
| κ·σ (conservation law) | {KAPPA_SIGMA:.1f} |

Tier: `FRAMEWORK` — proves structural shape, not empirical equivalence."""


def run_lift(choice: str) -> tuple[str, str, str, str]:
    src_cat, F = _REGISTRY[choice]
    witness = verify_methodology_functor(F)
    return (render_dag_md(src_cat), render_dag_md(_CARL),
            render_functor_md(F), render_witness_md(witness))


HEADER = """# Methodology Functor — live demo

A **methodology functor** lifts the procedure of a published paper onto a target domain
while preserving structure (composition, identity, node typing). When closure = 1.000,
every move in the source DAG has a structurally matching move in the target.

The source library encodes published research as typed DAGs (eight node types:
`PROBLEM → SUBSTRATE → COUPLING → CONTROL → MEASUREMENT → CHARACTERIZATION → WITNESS`,
plus `LIFT`). The target is **CARL** — the Coherence-Aware RL training paradigm.

Two source domains shown — quantum vortex hardware (Nature 2026) and classical
Kuramoto synchronization (1975). Both lift to CARL with closure 1.000.
"""

CTA = f"""---
## Want a lift for your paper?

If you have a paper or process you'd like encoded — methodology DAG extraction,
functor design into a target domain, closure verification — I take a small
number of personal consults each month.

[**→ Book a personal consult**]({PAYMENT_URL})

Built on [carl-studio](https://pypi.org/project/carl-studio/) by Intuition Labs LLC."""


with gr.Blocks(title="Methodology Functor — CARL") as demo:
    gr.Markdown(HEADER)

    choice = gr.Radio(
        choices=list(_REGISTRY.keys()),
        value="Vortex (Nature 2026)",
        label="Source methodology",
    )

    with gr.Row():
        with gr.Column(scale=1):
            src_dag = gr.Markdown()
        with gr.Column(scale=1):
            tgt_dag = gr.Markdown()

    with gr.Row():
        with gr.Column(scale=1):
            functor_md = gr.Markdown()
        with gr.Column(scale=1):
            witness_md = gr.Markdown()

    gr.Markdown(CTA)

    choice.change(fn=run_lift, inputs=choice,
                  outputs=[src_dag, tgt_dag, functor_md, witness_md])
    demo.load(fn=lambda: run_lift("Vortex (Nature 2026)"),
              outputs=[src_dag, tgt_dag, functor_md, witness_md])


if __name__ == "__main__":
    demo.launch()