"""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()