"""Operon Adaptive Assembly Dashboard — template selection, assembly, and experience pool.""" from __future__ import annotations from dataclasses import dataclass, field from typing import Any import gradio as gr # --------------------------------------------------------------------------- # Inline simulation types # --------------------------------------------------------------------------- @dataclass(frozen=True) class TaskFingerprint: task_shape: str tool_count: int subtask_count: int required_roles: tuple[str, ...] tags: tuple[str, ...] = () @dataclass(frozen=True) class PatternTemplate: template_id: str name: str topology: str stage_count: int fingerprint: TaskFingerprint tags: tuple[str, ...] = () success_rate: float | None = None @dataclass class ExperienceRecord: stage_name: str signal_category: str intervention_kind: str outcome_success: bool # --------------------------------------------------------------------------- # Preset library # --------------------------------------------------------------------------- TEMPLATES = [ PatternTemplate("t1", "Sequential Review Pipeline", "reviewer_gate", 2, TaskFingerprint("sequential", 2, 2, ("writer", "reviewer")), ("content", "review")), PatternTemplate("t2", "Enterprise Analysis Organism", "skill_organism", 3, TaskFingerprint("sequential", 4, 3, ("researcher", "strategist")), ("enterprise", "analysis")), PatternTemplate("t3", "Research Swarm", "specialist_swarm", 4, TaskFingerprint("parallel", 5, 4, ("analyst", "writer")), ("research",)), PatternTemplate("t4", "Single Worker", "single_worker", 1, TaskFingerprint("sequential", 1, 1, ("worker",)), ("simple",)), ] def _jaccard(a: tuple[str, ...], b: tuple[str, ...]) -> float: sa, sb = set(a), set(b) union = sa | sb return len(sa & sb) / len(union) if union else 1.0 def _score(tmpl: PatternTemplate, fp: TaskFingerprint) -> float: shape = 1.0 if tmpl.fingerprint.task_shape == fp.task_shape else 0.0 tool_prox = 1.0 / (1.0 + abs(tmpl.fingerprint.tool_count - fp.tool_count)) sub_prox = 1.0 / (1.0 + abs(tmpl.fingerprint.subtask_count - fp.subtask_count)) role_sim = _jaccard(tmpl.fingerprint.required_roles, fp.required_roles) tag_sim = _jaccard(tmpl.tags, fp.tags) sr = tmpl.success_rate if tmpl.success_rate is not None else 0.5 return 0.30 * shape + 0.15 * tool_prox + 0.15 * sub_prox + 0.20 * role_sim + 0.10 * tag_sim + 0.10 * sr # --------------------------------------------------------------------------- # Presets # --------------------------------------------------------------------------- PRESETS = { "Enterprise Analysis": TaskFingerprint("sequential", 3, 3, ("researcher", "strategist"), ("enterprise",)), "Content Review": TaskFingerprint("sequential", 2, 2, ("writer", "reviewer"), ("content",)), "Parallel Research": TaskFingerprint("parallel", 5, 4, ("analyst", "writer"), ("research",)), "Simple Task": TaskFingerprint("sequential", 1, 1, ("worker",), ("simple",)), } EXPERIENCE_PRESETS = { "Fresh (no experience)": [], "Warm (escalate works for epistemic)": [ ExperienceRecord("router", "epistemic", "escalate", True), ExperienceRecord("router", "epistemic", "escalate", True), ExperienceRecord("router", "epistemic", "retry", False), ], "Mixed (retry and escalate both tried)": [ ExperienceRecord("planner", "epistemic", "escalate", True), ExperienceRecord("planner", "epistemic", "retry", True), ExperienceRecord("planner", "somatic", "halt", True), ExperienceRecord("router", "epistemic", "retry", False), ], } # --------------------------------------------------------------------------- # Callbacks # --------------------------------------------------------------------------- def _run_selection(preset_name): if preset_name not in PRESETS: return "Select a preset." fp = PRESETS[preset_name] scored = [(t, round(_score(t, fp), 4)) for t in TEMPLATES] scored.sort(key=lambda x: x[1], reverse=True) rows = "" for i, (t, s) in enumerate(scored): bg = "#1a2a1a" if i == 0 else "transparent" badge = ' SELECTED' if i == 0 else "" rows += f""" {t.name}{badge} {t.topology} {t.stage_count} stages {s:.4f} """ fp_html = f"""
Fingerprint: shape={fp.task_shape}, tools={fp.tool_count}, subtasks={fp.subtask_count}, roles={fp.required_roles}, tags={fp.tags}
""" table = f"""{rows}
Template Topology Stages Score
""" return fp_html + table def _run_experience(exp_preset): if exp_preset not in EXPERIENCE_PRESETS: return "Select a preset." records = EXPERIENCE_PRESETS[exp_preset] if not records: return '

Empty experience pool — watcher uses rule-based decisions only.

' rows = "" for r in records: color = "#4a4" if r.outcome_success else "#a44" rows += f""" {r.stage_name} {r.signal_category} {r.intervention_kind} {'success' if r.outcome_success else 'failed'} """ # Compute recommendation success_by_kind: dict[str, int] = {} for r in records: if r.outcome_success: success_by_kind[r.intervention_kind] = success_by_kind.get(r.intervention_kind, 0) + 1 rec = max(success_by_kind, key=lambda k: success_by_kind[k]) if success_by_kind else "none" table = f"""{rows}
Stage Signal Intervention Outcome
""" rec_html = f'
Recommendation: {rec.upper()}
' return table + rec_html # --------------------------------------------------------------------------- # App # --------------------------------------------------------------------------- def build_app(): with gr.Blocks(title="Operon Adaptive Assembly", theme=gr.themes.Base()) as demo: gr.Markdown("# Operon Adaptive Assembly\nTemplate selection, organism construction, and experience-driven intervention.") with gr.Tab("Template Selection"): preset = gr.Dropdown(choices=list(PRESETS.keys()), label="Task Preset", value="Enterprise Analysis") btn = gr.Button("Select Template") out = gr.HTML() btn.click(_run_selection, inputs=[preset], outputs=[out]) with gr.Tab("Experience Pool"): exp_preset = gr.Dropdown(choices=list(EXPERIENCE_PRESETS.keys()), label="Experience Preset", value="Fresh (no experience)") btn2 = gr.Button("Show Pool & Recommendation") out2 = gr.HTML() btn2.click(_run_experience, inputs=[exp_preset], outputs=[out2]) return demo if __name__ == "__main__": app = build_app() app.launch()