Spaces:
Runtime error
Runtime error
| """Owner-only operator dashboard.""" | |
| from __future__ import annotations | |
| import os | |
| from typing import Dict | |
| import gradio as gr | |
| from jenaai.core.module import BaseModule | |
| class Module(BaseModule): | |
| """Operator dashboard with stream-safe toggle.""" | |
| def __init__(self, event_bus, config): | |
| super().__init__(event_bus, config) | |
| self.password = os.getenv("OPERATOR_PASSWORD", "change-me") | |
| self.stream_safe = True | |
| def _redact(self, value: str) -> str: | |
| if self.stream_safe: | |
| return "[REDACTED]" | |
| return value | |
| def _toggle(self, enabled: bool) -> str: | |
| self.stream_safe = enabled | |
| return "Stream-safe mode enabled" if enabled else "Full mode enabled" | |
| def _status(self) -> Dict[str, str]: | |
| return { | |
| "mode": "stream-safe" if self.stream_safe else "full", | |
| "prompt": self._redact("Prompt + memory context"), | |
| "logs": self._redact("Live console logs"), | |
| } | |
| async def launch(self) -> None: | |
| with gr.Blocks(title="JenaAI Operator") as app: | |
| gr.Markdown("# JenaAI Operator Dashboard") | |
| password = gr.Textbox(label="Password", type="password") | |
| mode = gr.Checkbox(label="Stream-safe mode", value=True) | |
| status = gr.JSON(label="Status") | |
| button = gr.Button("Load Status") | |
| def load_status(pwd: str, enabled: bool) -> Dict[str, str]: | |
| if pwd != self.password: | |
| return {"error": "Unauthorized"} | |
| self.stream_safe = enabled | |
| return self._status() | |
| button.click(load_status, inputs=[password, mode], outputs=status) | |
| app.queue() | |
| app.launch(server_name="0.0.0.0", server_port=7860, share=False) | |