""" Operon Plasmid Registry -- Horizontal Gene Transfer Demo ======================================================== Dynamic tool acquisition from a searchable registry with capability gating. Agents can discover, acquire, execute, and release tools at runtime -- like bacteria exchanging plasmids via horizontal gene transfer. Run locally: pip install gradio python space-plasmid/app.py Deploy to HuggingFace Spaces: Copy this directory to a new HF Space with sdk=gradio. """ import sys 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.core.types import Capability from operon_ai.organelles.mitochondria import Mitochondria from operon_ai.organelles.plasmid import Plasmid, PlasmidRegistry # --------------------------------------------------------------------------- # Preset capability sets # --------------------------------------------------------------------------- CAPABILITY_PRESETS: dict[str, set[Capability]] = { "No restrictions": set(), "READ_FS only": {Capability.READ_FS}, "NET only": {Capability.NET}, "NET + READ_FS": {Capability.NET, Capability.READ_FS}, "EXEC_CODE only": {Capability.EXEC_CODE}, "All capabilities": {Capability.NET, Capability.READ_FS, Capability.EXEC_CODE, Capability.WRITE_FS}, } # --------------------------------------------------------------------------- # Build default registry # --------------------------------------------------------------------------- def _build_registry() -> PlasmidRegistry: registry = PlasmidRegistry() registry.register(Plasmid( name="reverse", description="Reverse a string", func=lambda s: s[::-1], tags=frozenset({"text", "utility"}), version="1.0.0", )) registry.register(Plasmid( name="word_count", description="Count words in a string", func=lambda s: len(s.split()), tags=frozenset({"text", "analysis"}), version="1.0.0", )) registry.register(Plasmid( name="uppercase", description="Convert string to uppercase", func=lambda s: s.upper(), tags=frozenset({"text", "transform"}), version="1.0.0", )) registry.register(Plasmid( name="char_count", description="Count characters in a string", func=lambda s: len(s), tags=frozenset({"text", "analysis"}), version="1.2.0", )) registry.register(Plasmid( name="fetch_url", description="Fetch a URL (simulated)", func=lambda url: f"Content from {url}", tags=frozenset({"network", "io"}), version="1.0.0", required_capabilities=frozenset({Capability.NET}), )) registry.register(Plasmid( name="read_file", description="Read a file (simulated)", func=lambda path: f"Contents of {path}", tags=frozenset({"filesystem", "io"}), version="1.0.0", required_capabilities=frozenset({Capability.READ_FS}), )) registry.register(Plasmid( name="run_script", description="Execute a script (simulated)", func=lambda code: f"Executed: {code}", tags=frozenset({"execution", "code"}), version="2.0.0", required_capabilities=frozenset({Capability.EXEC_CODE}), )) return registry # --------------------------------------------------------------------------- # Session state # --------------------------------------------------------------------------- _registry = _build_registry() _mito: Mitochondria | None = None _event_log: list[str] = [] def _reset_agent(caps_label: str) -> str: global _mito, _event_log _event_log = [] if caps_label == "No restrictions": _mito = Mitochondria(silent=True) _event_log.append("Agent created with NO capability restrictions.") else: caps = CAPABILITY_PRESETS.get(caps_label, set()) _mito = Mitochondria(allowed_capabilities=caps, silent=True) cap_names = sorted(c.value for c in caps) if caps else ["none"] _event_log.append(f"Agent created with capabilities: {cap_names}") return _render_state() def _render_state() -> str: if _mito is None: return '
No agent created yet. Select capabilities and click Create.
' tools = list(_mito.tools.keys()) tool_badges = "" if tools: tool_badges = " ".join( f'{t}' for t in tools ) else: tool_badges = 'none' caps = _mito.allowed_capabilities if caps is None: cap_text = "unrestricted" elif caps: cap_text = ", ".join(sorted(c.value for c in caps)) else: cap_text = "none (empty set)" html = ( f'
' f'
Agent State
' f'
Capabilities: {cap_text}
' f'
Tools ({len(tools)}): {tool_badges}
' f'
ROS: {_mito.get_ros_level():.2f} | ' f'Operations: {_mito._operations_count}
' f'
' ) return html def _render_log() -> str: if not _event_log: return "" rows = "" for i, entry in enumerate(reversed(_event_log[-20:]), 1): bg = "#f0fdf4" if "OK" in entry or "Success" in entry else ( "#fef2f2" if "BLOCKED" in entry or "Error" in entry or "Failed" in entry else "#f9fafb" ) rows += f'
{entry}
' return f'
{rows}
' # --------------------------------------------------------------------------- # Registry browsing # --------------------------------------------------------------------------- def browse_registry(search_query: str) -> str: if search_query.strip(): results = _registry.search(search_query) else: results = [_registry.get(item["name"]) for item in _registry.list_available()] if not results: return '
No plasmids found.
' rows = "" for p in results: caps = sorted(c.value for c in p.required_capabilities) if p.required_capabilities else ["none"] tags = sorted(p.tags) cap_color = "#22c55e" if caps == ["none"] else "#ef4444" rows += ( f'' f'{p.name}' f'{p.description}' f'{p.version}' f'{", ".join(tags)}' f'{", ".join(caps)}' f'' ) return ( f'' f'' f'' f'' f'' f'' f'{rows}
NameDescriptionVersionTagsRequired Caps
' ) # --------------------------------------------------------------------------- # Acquire / Release / Execute # --------------------------------------------------------------------------- def acquire_plasmid(plasmid_name: str) -> tuple[str, str]: if _mito is None: _event_log.append("Error: Create an agent first.") return _render_state(), _render_log() name = plasmid_name.strip() if not name: _event_log.append("Error: Enter a plasmid name.") return _render_state(), _render_log() result = _mito.acquire(name, _registry) if result.success: _event_log.append(f"OK -- Acquired '{name}' successfully.") else: _event_log.append(f"BLOCKED -- Acquire '{name}' failed: {result.error}") return _render_state(), _render_log() def release_plasmid(plasmid_name: str) -> tuple[str, str]: if _mito is None: _event_log.append("Error: Create an agent first.") return _render_state(), _render_log() name = plasmid_name.strip() if not name: _event_log.append("Error: Enter a tool name to release.") return _render_state(), _render_log() try: _mito.release(name) _event_log.append(f"OK -- Released '{name}' (plasmid curing).") except ValueError as e: _event_log.append(f"Failed -- Release '{name}': {e}") return _render_state(), _render_log() def execute_tool(expression: str) -> tuple[str, str, str]: if _mito is None: _event_log.append("Error: Create an agent first.") return "", _render_state(), _render_log() expr = expression.strip() if not expr: return "", _render_state(), _render_log() result = _mito.metabolize(expr) if result.success and result.atp: result_html = ( f'
' f'Result: ' f'{result.atp.value!r}' f'
' ) _event_log.append(f"OK -- `{expr}` = {result.atp.value!r}") else: result_html = ( f'
' f'Error: ' f'{result.error}' f'
' ) _event_log.append(f"Failed -- `{expr}`: {result.error}") return result_html, _render_state(), _render_log() # --------------------------------------------------------------------------- # Full scenario demo # --------------------------------------------------------------------------- def run_scenario() -> tuple[str, str]: global _mito, _event_log _event_log = [] lines: list[str] = [] lines.append("### Scenario: Capability-Gated Tool Acquisition\n") # Step 1: Create restricted agent _mito = Mitochondria(allowed_capabilities={Capability.READ_FS}, silent=True) _event_log.append("Agent created with capabilities: ['read_fs']") lines.append("**Step 1:** Create agent with `READ_FS` capability only.\n") # Step 2: Acquire safe tools r1 = _mito.acquire("reverse", _registry) _event_log.append(f"Acquire 'reverse': {'OK' if r1.success else 'BLOCKED -- ' + str(r1.error)}") lines.append(f"**Step 2:** Acquire `reverse` (no caps required): **{'Success' if r1.success else 'Blocked'}**\n") r2 = _mito.acquire("word_count", _registry) _event_log.append(f"Acquire 'word_count': {'OK' if r2.success else 'BLOCKED -- ' + str(r2.error)}") lines.append(f"**Step 3:** Acquire `word_count` (no caps required): **{'Success' if r2.success else 'Blocked'}**\n") # Step 3: Try to acquire NET tool r3 = _mito.acquire("fetch_url", _registry) _event_log.append(f"Acquire 'fetch_url': {'OK' if r3.success else 'BLOCKED -- ' + str(r3.error)}") lines.append(f"**Step 4:** Acquire `fetch_url` (requires `NET`): **{'Success' if r3.success else 'Blocked'}**\n") if not r3.success: lines.append(f"> *{r3.error}*\n") # Step 4: Acquire FS tool (allowed) r4 = _mito.acquire("read_file", _registry) _event_log.append(f"Acquire 'read_file': {'OK' if r4.success else 'BLOCKED -- ' + str(r4.error)}") lines.append(f"**Step 5:** Acquire `read_file` (requires `READ_FS`): **{'Success' if r4.success else 'Blocked'}**\n") # Step 5: Execute acquired tools lines.append("**Step 6:** Execute acquired tools:\n") res = _mito.metabolize('reverse("hello world")') val = res.atp.value if res.success and res.atp else res.error _event_log.append(f"Execute reverse('hello world'): {val}") lines.append(f"- `reverse(\"hello world\")` = `{val}`\n") res = _mito.metabolize('word_count("the quick brown fox jumps")') val = res.atp.value if res.success and res.atp else res.error _event_log.append(f"Execute word_count: {val}") lines.append(f"- `word_count(\"the quick brown fox jumps\")` = `{val}`\n") # Step 6: Release and verify _mito.release("reverse") _event_log.append("OK -- Released 'reverse' (plasmid curing).") lines.append("**Step 7:** Release `reverse` (plasmid curing).\n") res = _mito.metabolize('reverse("test")') _event_log.append(f"Execute after release: {'Success' if res.success else 'Failed -- ' + str(res.error)}") lines.append(f"**Step 8:** Execute `reverse(\"test\")` after release: **{'Success' if res.success else 'Blocked'}** (tool removed)\n") # Step 7: Re-acquire r5 = _mito.acquire("reverse", _registry) _event_log.append(f"Re-acquire 'reverse': {'OK' if r5.success else 'BLOCKED'}") lines.append(f"**Step 9:** Re-acquire `reverse`: **{'Success' if r5.success else 'Blocked'}**\n") lines.append(f"\n**Final tools:** {list(_mito.tools.keys())}") return "\n".join(lines), _render_log() # --------------------------------------------------------------------------- # Gradio UI # --------------------------------------------------------------------------- def build_app() -> gr.Blocks: with gr.Blocks(title="Operon Plasmid Registry") as app: gr.Markdown( "# Operon Plasmid Registry -- Horizontal Gene Transfer\n" "Dynamic tool acquisition with optional capability gating (Paper §6.2). " "Agents discover and absorb tools from a registry. When a capability " "envelope is configured, acquisition checks reject tools that exceed it.\n\n" "[GitHub](https://github.com/coredipper/operon) | " "[Paper](https://github.com/coredipper/operon/tree/main/article)" ) with gr.Tabs(): # ── Tab 1: Interactive ──────────────────────────────── with gr.TabItem("Interactive"): with gr.Row(): with gr.Column(scale=1): gr.Markdown("### 1. Create Agent") caps_dropdown = gr.Dropdown( choices=list(CAPABILITY_PRESETS.keys()), value="READ_FS only", label="Allowed Capabilities", ) create_btn = gr.Button("Create Agent", variant="primary") agent_state = gr.HTML(label="Agent State") with gr.Column(scale=2): gr.Markdown("### 2. Registry") search_input = gr.Textbox( label="Search plasmids", placeholder="e.g. text, reverse, network...", ) registry_html = gr.HTML(value=browse_registry("")) with gr.Row(): with gr.Column(): gr.Markdown("### 3. Acquire / Release") plasmid_input = gr.Textbox( label="Plasmid name", placeholder="e.g. reverse", ) with gr.Row(): acquire_btn = gr.Button("Acquire", variant="primary") release_btn = gr.Button("Release", variant="secondary") with gr.Column(): gr.Markdown("### 4. Execute") expr_input = gr.Textbox( label="Expression", placeholder='e.g. reverse("hello")', ) exec_btn = gr.Button("Execute", variant="primary") exec_result = gr.HTML() gr.Markdown("### Event Log") log_html = gr.HTML() # Bindings create_btn.click(fn=_reset_agent, inputs=[caps_dropdown], outputs=[agent_state]) search_input.change(fn=browse_registry, inputs=[search_input], outputs=[registry_html]) search_input.submit(fn=browse_registry, inputs=[search_input], outputs=[registry_html]) acquire_btn.click(fn=acquire_plasmid, inputs=[plasmid_input], outputs=[agent_state, log_html]) release_btn.click(fn=release_plasmid, inputs=[plasmid_input], outputs=[agent_state, log_html]) exec_btn.click(fn=execute_tool, inputs=[expr_input], outputs=[exec_result, agent_state, log_html]) expr_input.submit(fn=execute_tool, inputs=[expr_input], outputs=[exec_result, agent_state, log_html]) # ── Tab 2: Full scenario ────────────────────────────── with gr.TabItem("Guided Scenario"): gr.Markdown( "### Capability-Gated Tool Acquisition\n\n" "Run a complete scenario demonstrating:\n" "- Agent creation with limited capabilities\n" "- Safe tool acquisition (no caps required)\n" "- Blocked acquisition (insufficient capabilities)\n" "- Tool execution, release (plasmid curing), and re-acquisition" ) scenario_btn = gr.Button("Run Scenario", variant="primary") scenario_md = gr.Markdown() scenario_log = gr.HTML() scenario_btn.click(fn=run_scenario, outputs=[scenario_md, scenario_log]) return app if __name__ == "__main__": app = build_app() app.launch(theme=gr.themes.Soft())