Spaces:
Runtime error
Runtime error
File size: 1,779 Bytes
09f1b19 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 | """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)
|