"""HTML rendering helpers for the Gradio command center."""
from __future__ import annotations
import os
import base64
from html import escape
from pathlib import Path
from .catalog import catalog_summary, parameter_budget
from .schema import GenerationRun
def badge(label: str, tone: str = "neutral") -> str:
"""
Generate an HTML badge element with a tone-specific CSS class.
Parameters:
label (str): The badge text content.
tone (str): A CSS class modifier for the badge style; defaults to "neutral".
Returns:
str: An HTML span element with the specified tone class.
"""
return f'{escape(label)}'
def icon(name: str) -> str:
paths = {
"forge": '',
"wardrobe": '',
"lore": '',
"models": '',
"security": '',
"runs": '',
"zoom": '',
"frame": '',
"lock": '',
"dot": '',
}
return f''
def _metric(label: str, value: str, tone: str = "neutral") -> str:
"""
Render an HTML metric block with a label and value.
Parameters:
tone (str): The styling variant name (default: "neutral").
Returns:
str: HTML markup for the metric.
"""
return f'
{escape(label)}{escape(value)}
'
def _display_state(value: str) -> str:
"""Return user-facing state labels without changing machine state values."""
normalized = str(value or "").replace("_", " ").strip().lower()
labels = {
"blocked": "Export Gate Active",
"generated": "Generated",
"missing secret": "Optional - Secret Required",
"missing_secret": "Optional - Secret Required",
"export_ready": "Export Ready",
"configured": "CONFIGURED",
"dry-run": "Dry Run",
"dry run": "Dry Run",
"success": "SUCCESS",
}
return labels.get(normalized, normalized.title() if normalized else "Pending")
def _gate_label(export_gate: str) -> str:
normalized = str(export_gate or "pending").upper()
if normalized == "BLOCKED":
return "Export Blocked - Override Available"
if normalized == "CLEAR":
return "EXPORT CLEAR"
if normalized == "PENDING":
return "Export waits for review"
return f"EXPORT {normalized}"
def _env_configured(*names: str) -> bool:
"""
Determine whether any of the provided environment variables are set and truthy.
Returns:
True if any of the environment variables exists and is truthy, False otherwise.
"""
return any(bool(os.environ.get(name)) for name in names)
def _provider_configured(base_url_name: str, *key_names: str) -> bool:
"""
Determines if a provider is configured with both a base URL and authentication keys.
Returns:
True if the base URL environment variable is set and at least one of the
provided key environment variables is set, False otherwise.
"""
return bool(os.environ.get(base_url_name)) and _env_configured(*key_names)
_SCAN_REDACTION_TERMS = (
"payload",
"payload_excerpt",
"hidden content",
"recovered",
"raw byte",
"base64",
"hex dump",
)
def _redact_scan_text(value: object) -> str:
"""
Sanitizes scan text by redacting sensitive content and truncating excessively long values.
Parameters:
value (object): A scan detail to sanitize.
Returns:
str: A redaction notice if the text contains sensitive terms; the original text truncated
to 177 characters plus "..." if it exceeds 180 characters; otherwise the original
text converted to a string.
"""
text = str(value)
lowered = text.lower()
if any(term in lowered for term in _SCAN_REDACTION_TERMS):
return "Redacted scan detail; review the gated ST3GG evidence packet."
if len(text) > 180:
return text[:177] + "..."
return text
def _space_runtime_status() -> dict[str, str]:
"""
Determines the current runtime environment and active provider configuration.
Returns:
dict[str, str]: A dict with `space_id`, `hardware`, `bucket`, and `secrets` keys describing the deployment context and configured provider secrets.
"""
space_id = os.environ.get("SPACE_ID") or os.environ.get("HF_SPACE_ID") or "local-preview"
hardware = os.environ.get("SPACE_HARDWARE") or os.environ.get("NEXUS_SPACE_HARDWARE") or "ZeroGPU"
bucket = "/data mounted" if os.path.isdir("/data") else "bucket optional"
configured = []
if _env_configured("HF_TOKEN", "HUGGING_FACE_HUB_TOKEN"):
configured.append("HF")
if _provider_configured("MINICPM_BASE_URL", "MINICPM_API_KEY", "OPENBMB_API_KEY"):
configured.append("MiniCPM")
if _provider_configured("NEMOTRON_BASE_URL", "NEMOTRON_API_KEY", "NVIDIA_API_KEY"):
configured.append("Nemotron")
if _env_configured("FAL_KEY", "NETLIFY_AUTH_TOKEN", "OPENAI_API_KEY", "MODAL_TOKEN_ID"):
configured.append("optional")
secrets = "configured: " + ", ".join(configured) if configured else "no provider secrets"
return {
"space_id": space_id,
"hardware": hardware,
"bucket": bucket,
"secrets": secrets,
}
def _image_data_uri(path: str | None) -> str | None:
"""
Create a data URI string from a file's contents, selecting MIME type by extension.
Returns:
data_uri (str | None): A data URI string with base64-encoded file contents if the path exists and is readable; `None` otherwise. MIME type is determined by file extension (.png, .jpg/.jpeg, .webp; otherwise application/octet-stream).
"""
if not path:
return None
target = Path(path)
if not target.exists() or not target.is_file():
return None
suffix = target.suffix.lower()
mime = "image/png" if suffix == ".png" else "image/jpeg" if suffix in {".jpg", ".jpeg"} else "image/webp" if suffix == ".webp" else "application/octet-stream"
data = base64.b64encode(target.read_bytes()).decode("ascii")
return f"data:{mime};base64,{data}"
def render_command_header() -> str:
"""
Render the command header section of the dashboard.
Returns:
An HTML string containing the command header, description, and status badges.
"""
return f"""
COMMAND INPUT
Raven Chronicle Active WeavePrompt, reference scan, model route, and checkpoint controls stay in one sticky operator strip.
"""
def render_workflow(run: GenerationRun | None = None, operator_state: dict | None = None) -> str:
"""
Render an SVG workflow diagram and console showing the generation pipeline.
The diagram visualizes a multi-stage workflow (seed prompt, refine, judge, locate, generate, video, checkpoint)
with nodes, edges, and status indicators. Node content and labels reflect the provided run and operator context.
Parameters:
run: GenerationRun object providing run-specific values for trust score, checkpoint ID, model labels,
video preset, and required actions. Defaults to mock values if None.
operator_state: Dict containing operator context including checkpoint status, provider state, recommendation,
and message. Defaults to an empty dict.
Returns:
str: HTML section containing SVG workflow graph, legend badges, and console cards with checkpoint,
action, model lane, and signal information.
"""
operator_state = operator_state or {}
score = run.checkpoint.trust_score if run else 0.82
checkpoint_id = run.checkpoint.checkpoint_id if run else "nw-dry-run"
checkpoint_status = str(operator_state.get("checkpoint", "pending_review" if run else "pending"))
provider_state = str(operator_state.get("provider_state", "dry-run" if run else "idle"))
recommendation = operator_state.get("message") or (run.checkpoint.recommendation.replace("_", " ").title() if run else "Awaiting Run")
required_actions = run.checkpoint.required_actions if run else ["Review candidate thumbnails before promotion"]
action_label = required_actions[0] if required_actions else "No action pending"
if checkpoint_status == "approved":
action_label = "Checkpoint approved; export packet may be prepared after ST3GG gate."
elif provider_state == "stopped":
action_label = "Provider handoff stopped; dry-run evidence remains available."
model_label = _short_repo(run.model_stack[0].repo_id) if run and run.model_stack else "FLUX.2"
locate_label = run.inspection.locate_model.split("/")[-1] if run else "LocateAnything-3B"
nodes = {
"seed": (35, 52, 190, 210, "Seed Prompt", ["Rogue archivist moving", "through rain-slick neon", "city, couture layers."], "Text-to-Image (FLUX.2)", "complete", "red"),
"refine": (275, 52, 185, 160, "Refine", ["Prompt Refiner", "Style Harmonizer", "Negative Purge"], "Qwen2.5-7B", "complete", "violet"),
"judge": (540, 52, 185, 160, "Judge", ["Aesthetic Scorer", "ST3GG Policy Filter", f"Score {score:.2f}"], "MiniCPM / Nemotron", "complete", "blue"),
"locate": (785, 52, 185, 160, "Locate", ["Reference Locator", "Pose & Composition", "IP-Adapter"], "Refs 3/5", "complete", "cyan"),
"generate": (275, 280, 235, 210, "Generate", ["Image / Video Generation", "FLUX.2 9B + adapter stack", "4B sidecar fallback"], "Steps 4 CFG 1.0", "ready", "green"),
"video": (590, 280, 235, 210, "Video Path", ["Image to Video", "Frame interpolation", run.video.preset if run else "Wan2.2 / LTX swap"], "Duration 5.6s 24fps", "ready", "blue"),
"checkpoint": (880, 285, 185, 185, "Human Checkpoint", ["Human review required", "Verify intent, vibe,", "and output before final."], "Review Now", "paused", "amber"),
}
edges = [
("seed", "refine"), ("refine", "judge"), ("judge", "locate"), ("locate", "video"),
("refine", "generate"), ("generate", "video"), ("video", "checkpoint"), ("judge", "checkpoint"),
]
lines = []
for source, target in edges:
x1, y1, w1, h1, *_ = nodes[source]
x2, y2, w2, h2, *_ = nodes[target]
y_offset = 76 if source in {"seed", "refine", "judge", "locate"} else 96
lines.append(
f''
)
cards = []
for node_id, (x, y, width, height, title, body, footer, status, tone) in nodes.items():
body_lines = "".join(f'{escape(line)}' for idx, line in enumerate(body))
thumb_strip = ""
if node_id in {"generate", "video"}:
thumbs = []
for idx in range(4):
tx = x + 16 + idx * 47
thumbs.append(
f''
)
thumb_strip = "".join(thumbs)
cards.append(
f"""
{escape(title)}
{body_lines}
{thumb_strip}
{escape(footer)}
"""
)
generation = operator_state.get("generation") or {}
generated = isinstance(generation, dict) and generation.get("status") == "success"
output_summary = "Real FLUX.2 Klein 9B artifact generated" if generated else "Awaiting generated artifact"
checkpoint_summary = _display_state(checkpoint_status)
export_summary = _display_state(str(operator_state.get("export", provider_state)))
next_summary = "Add override reason or complete review" if provider_state == "blocked" else action_label
return f"""
Active Weave Live / Weave ID: 4f7c9e2b
Layout Auto{icon("zoom")}{icon("frame")}
Current Run SummaryOutput{escape(output_summary)}SecurityST3GG review activeCheckpoint{escape(checkpoint_summary)}Export{escape(export_summary)}Next{escape(next_summary)}
{badge("Text Flow", "accent")} {badge("Refine Loop", "violet")} {badge("Policy Gate", "blue")} {badge("Media Flow", "cyan")} {badge("Human Gate", "warn")} {badge(_display_state(provider_state), "pass" if provider_state == "exported" else "warn" if provider_state in {"blocked", "stopped"} else "muted")}
Selected Node
Human Checkpoint{escape(str(recommendation))} / {escape(checkpoint_id)}
Next Operator Action
{escape(action_label)}Checkpoint blocks video promotion until reviewed.
Pinned Model Lanes
{escape(model_label)} + {escape(locate_label)} + ST3GGCore lanes stay fixed; helper lanes may rotate.
Hackathon Signal
Workflow, governance, visual creationJudge view keeps product purpose visible without a landing page.
"""
def render_artifact_lane(run: GenerationRun | None = None, scan: dict | None = None, operator_state: dict | None = None) -> str:
"""
Render an artifact preview panel with generated output and status cards.
Displays the current artifact output (if available from a provider call), preview metadata
about checkpoint and export status, and a grid of artifact cards showing progress through
generation stages. Supports both dry-run and live provider output scenarios.
Parameters:
run (GenerationRun | None): Generation run with output image path, refined prompt, and continuity data.
If None, shows dry-run placeholder content.
scan (dict | None): Security scan state with status and export gate information.
Defaults to {"status": "idle", "export_gate": "pending"}.
operator_state (dict | None): Operator context including checkpoint status, provider state, and generated output.
Defaults to {}.
Returns:
str: HTML markup for the artifact preview section.
"""
scan = scan or {"status": "idle", "export_gate": "pending"}
operator_state = operator_state or {}
active_prompt = run.refined_prompt.refined[:150] if run else "Describe the look and click Generate Image."
checkpoint = operator_state.get("checkpoint", getattr(run.checkpoint, "recommendation", "pending") if run else "pending")
provider_state = str(operator_state.get("provider_state", "dry-run" if run else "idle"))
generation = operator_state.get("generation") or {}
generated_uri = _image_data_uri(generation.get("output_path")) if isinstance(generation, dict) else None
generated_status = str(generation.get("status", "")) if isinstance(generation, dict) else ""
generated_message = str(generation.get("message", "")) if isinstance(generation, dict) else ""
output_ready = bool(generated_uri)
preview_mode = {
"idle": "Idle",
"dry-run": "Dry Run",
"checkpointed": "Checkpointed",
"export_ready": "Export Ready",
"exported": "Exported",
"blocked": "Export Gate Active",
"stopped": "Stopped",
"generated": "Generated",
}.get(provider_state, provider_state.replace("_", " ").title())
demo_seed = (run.checkpoint.checkpoint_id[-4:] if run else "0000").upper()
export_gate = str(scan.get("export_gate", "pending")).upper()
checkpoint_label = str(checkpoint).replace("_", " ").title()
next_action = "Generate an image to unlock checkpoint review."
if output_ready and str(checkpoint) != "approved":
next_action = "Approve checkpoint to unlock audit export."
elif output_ready and str(checkpoint) == "approved" and export_gate != "CLEAR":
next_action = "Add override reason or complete ST3GG review, then prepare audit export."
elif output_ready and str(checkpoint) == "approved":
next_action = "Prepare audit export."
return f"""
{'Generated image ready' if output_ready else 'Output'}Real output and export evidence appear here
{badge(_gate_label(export_gate), "warn" if export_gate == "BLOCKED" else "pass" if export_gate == "CLEAR" else "muted")}
{'' if generated_uri else ''}
PRIMARY OUTPUT STAGE / {escape(generated_status.upper() or "JUDGE-SAFE DEMO OUTPUT")} / SEED {escape(demo_seed)}
{'Real FLUX.2 Klein artifact' if generated_uri else 'No image generated yet'}{escape(generated_message or active_prompt)}
Output{'Ready' if output_ready else 'Waiting'}
Checkpoint{escape(checkpoint_label)}
Export{escape(_gate_label(export_gate))}
Current state{escape(preview_mode)}
SecurityST3GG review active
Next action{escape(next_action)}
"""
def render_operations_panel(
active_section: str = "Forge",
run: GenerationRun | None = None,
scan: dict | None = None,
relay_status: dict | None = None,
*,
adult_mode: bool = False,
operator_state: dict | None = None,
) -> str:
"""
Render an operations panel with section-specific operation cards.
Generates an HTML section containing three operation cards tailored to the selected section,
with content drawn from run state, scan results, relay decisions, and operator context.
Invalid section names default to "Forge".
Parameters:
active_section (str): The section name determining which operation cards to display.
Valid sections are "Forge", "Wardrobe", "Lore", "Models", "Security", and "Runs".
adult_mode (bool): If True, scope label is "Private research scope"; otherwise "Public demo scope".
operator_state (dict | None): Operator context dict with "provider_state" and "message" keys.
Returns:
str: HTML string representing the operations panel section.
"""
scan = scan or {"status": "idle", "export_gate": "pending", "findings": []}
relay_status = relay_status or {}
operator_state = operator_state or {}
section = active_section if active_section in {"Forge", "Wardrobe", "Lore", "Models", "Security", "Runs"} else "Forge"
checkpoint = getattr(run, "checkpoint", None) if run else None
outfit = getattr(run, "outfit", None) if run else None
lore = getattr(run, "lore", None) if run else None
run_id = getattr(checkpoint, "checkpoint_id", "not-started")
outfit_count = len(getattr(outfit, "slots", []) or [])
lore_count = len(getattr(lore, "beats", []) or [])
scan_status = str(scan.get("status", "idle")).upper()
export_gate = str(scan.get("export_gate", "pending")).upper()
provider_state = str(operator_state.get("provider_state", "idle")).replace("_", " ").upper()
operator_message = str(operator_state.get("message", "No operator action yet."))
decisions = relay_status.get("decisions", [])
first_decision = decisions[0] if decisions else {}
first_primary = (first_decision.get("primary") or {}) if first_decision else {}
adult_scope = "Private research scope" if adult_mode else "Public demo scope"
panels = {
"Forge": [
("Prompt contract", "Taste-refined prompt, material locks, negative purge, and checkpoint requirements."),
("Active run", f"{run_id} / checkpoint remains human-reviewed before video promotion."),
("Provider posture", f"{provider_state}: {operator_message}"),
],
"Wardrobe": [
("Slot coverage", f"{outfit_count or 9} garment/accessory regions tracked with locks and edit priority."),
("Footwear focus", "Platform boots, stilettos, high-heel boots, hardware, and silhouette constraints stay first-class."),
("Locate map", "Reference regions feed preflight and post-generation outfit verification."),
],
"Lore": [
("Beat budget", f"{lore_count or 6} compact beats: identity, garment meaning, world context, emotion, motion."),
("Video checkpoint", "Video presets remain handoff plans until human approval."),
("Continuity locks", "Lore-to-video keeps garment meaning and motion cue visible without tab sprawl."),
],
"Models": [
("Primary helper", _short_repo(str(first_primary.get("repo_id", "pending")))),
("Rotation mode", "Pinned core stays fixed; helper lanes rotate by license, budget, quota, and health."),
("Scope", adult_scope),
],
"Security": [
("ST3GG state", f"{scan_status} / export {export_gate}"),
(
"Findings",
"; ".join(_redact_scan_text(item) for item in (scan.get("findings") or [])[:2])
or ("No findings." if scan_status != "IDLE" else "No upload selected."),
),
("Public export", "Consent, provenance, metadata, age, dataset, and payload gates stay active."),
],
"Runs": [
("Current checkpoint", run_id),
("Ledger mode", f"Operator state: {provider_state}. Run JSON, catalog summary, and ST3GG evidence remain in the evidence accordion."),
("Rollback path", "Feature branches and draft PRs carry implementation checkpoints."),
],
}
rows = "".join(
f"""
{escape(title)}
{escape(body)}
"""
for title, body in panels[section]
)
return f"""
{escape(section)} OperationsSection-aware control surface for the selected command rail lane
"""
def _short_repo(repo_id: str) -> str:
"""
Extract the repository name from a repository identifier.
Returns the last segment after splitting on forward slashes.
Returns:
str: The repository name
"""
return repo_id.split("/")[-1]
def _render_relay_panel(relay_status: dict | None = None) -> str:
"""
Render an HTML panel summarizing model relay decisions and pinned lanes.
Returns:
str: An HTML string containing the GMR panel with pinned core lanes, up to two decisions, and dedup metrics.
"""
relay_status = relay_status or {}
pinned = relay_status.get("pinned", {})
decisions = relay_status.get("decisions", [])
pinned_labels = []
for lane in ["image_generation", "grounding", "security"]:
record = pinned.get(lane)
if record:
pinned_labels.append(_short_repo(record["repo_id"]))
pinned_rows = [
f"""
pinned core{escape(" / ".join(pinned_labels[:3]) or "pending")}image, grounding, and security never rotate
"""
]
decision_rows = []
for decision in decisions[:2]:
primary = decision.get("primary") or {}
fallbacks = decision.get("fallbacks") or []
fallback_label = ", ".join(_short_repo(item["repo_id"]) for item in fallbacks[:2]) or "none"
decision_rows.append(
f"""
"""
def render_provider_cards(relay_status: dict | None = None, adult_mode: bool = False) -> str:
"""
Generates an HTML panel displaying provider cards with operational status and licensing information.
Constructs cards for each relay-decided provider showing the lane, repository, provider name, license gate status, and health meter. Includes optional provider cards for off-by-default gateways. The panel header shows the current operational mode (private research or public demo).
Parameters:
relay_status (dict | None): Relay status containing provider decisions with quota impact and license gate data. If None, defaults to empty dict.
adult_mode (bool): If True, displays "PRIVATE RESEARCH" mode label; otherwise displays "PUBLIC DEMO SAFE".
Returns:
str: HTML section markup for the provider cards panel.
"""
relay_status = relay_status or {}
decisions = relay_status.get("decisions", [])
optional_statuses = {
"openbmb": "configured" if _provider_configured("MINICPM_BASE_URL", "MINICPM_API_KEY", "OPENBMB_API_KEY") else "missing secret",
"nvidia": "configured" if _provider_configured("NEMOTRON_BASE_URL", "NEMOTRON_API_KEY", "NVIDIA_API_KEY") else "missing secret",
"fal": "configured" if _env_configured("FAL_KEY") else "deferred",
"netlify": "configured" if _env_configured("NETLIFY_AUTH_TOKEN", "NETLIFY_SITE_ID", "OPENAI_BASE_URL") else "deferred",
"cloudflare": "configured" if _env_configured("CLOUDFLARE_API_TOKEN", "CF_ACCOUNT_ID") else "deferred",
}
cards = []
for decision in decisions[:5]:
primary = decision.get("primary") or {}
quota = decision.get("quota_impact") or {}
provider = primary.get("provider", "blocked")
repo = _short_repo(primary.get("repo_id", "blocked"))
lane = decision.get("lane", "helper").replace("_", " ")
status = quota.get("status", "blocked")
provider_state = "dry-run" if status == "ready" else "blocked" if status == "blocked" else "limited"
if provider in optional_statuses:
provider_state = optional_statuses[provider]
tone = "pass" if provider_state == "configured" else "warn" if provider_state in {"limited", "failed"} else "muted"
gate = primary.get("license_gate", "unknown")
cards.append(
f"""
'.format(badge("DRY-RUN", "muted")))
for provider, state in optional_statuses.items():
cards.append(
f"""
optional gateway
{escape(provider.title())}optional lane / secrets required
{badge(_display_state(state), "pass" if state == "configured" else "muted")}{badge("SPONSOR LANE" if provider in {"openbmb", "nvidia", "hf_nvidia"} else "DEFERRED", "muted")}
Optional Provider LanesVisible capability cards; success is claimed only after configured calls
{badge(mode_label.upper(), "warn" if adult_mode else "pass")}
{"".join(cards)}
"""
def _scan_status_tone(scan_status: str) -> str:
"""
Map a scan status to a display tone.
Returns:
str: "pass" if status is "pass", "warn" if status is "review" or "error", "muted" otherwise.
"""
if scan_status == "pass":
return "pass"
if scan_status in {"review", "error"}:
return "warn"
return "muted"
def render_inspector(
run: GenerationRun | None = None,
scan: dict | None = None,
relay_status: dict | None = None,
operator_state: dict | None = None,
) -> str:
"""
Render an inspector sidebar with taste profile, material checklist, model stack, sponsor evidence, and ST3GG scan details.
Returns:
str: HTML markup for the inspector panel.
"""
if run:
checks = [
("Patent Leather", True),
("Faux Fur", any(slot.material == "faux_fur" for slot in run.outfit.slots)),
("Lace / Mesh", any("lace" in slot.material for slot in run.outfit.slots)),
("Crimson Hardware", any(slot.material == "crimson_hardware" for slot in run.outfit.slots)),
("Platform Boots", any(slot.name == "footwear" for slot in run.outfit.slots)),
("Layered Garments", True),
]
stack_label = " / ".join(_short_repo(model.repo_id) for model in run.model_stack[:3])
model_rows = f"
' if generated_uri else ''}