Compare commits

...

6 Commits

Author SHA1 Message Date
Svrnty
f59b81b9e7 Render Runtime Health detail surface
Some checks failed
plugin-tests / test (push) Failing after 6s
upstream-drift / drift (push) Failing after 2s
2026-05-29 06:02:08 -04:00
Svrnty
114ffa4067 Align Runtime Health with canonical envelope 2026-05-29 03:35:20 -04:00
Svrnty
3685710fe8 Mount Cortex OS Runtime Health as WebUI panel 2026-05-29 03:13:56 -04:00
Svrnty
1707a7b09d Extract Cortex OS Hermes WebUI Host Adapter 2026-05-29 02:53:40 -04:00
Svrnty
8b6c810f4a Add Canvas command surface 2026-05-28 21:44:02 -04:00
Svrnty
cf723141a4 Implement S23 runtime health slice 2026-05-28 21:38:50 -04:00
24 changed files with 3785 additions and 15 deletions

View File

@ -2,7 +2,7 @@
**Upstream version:** v0.51.118
**Plugin version:** 0.5.0
**Total dependencies:** 34 (25 public API · 0 forced internal · 9 frontend)
**Total dependencies:** 62 (41 public API · 0 forced internal · 21 frontend)
> **Auto-generated by `scripts/ast-connection-map.py`. Do not hand-edit.**
> To change a justification, edit the `# CONNECTION:` comment above the
@ -14,23 +14,39 @@
| Plugin location | Upstream symbol | Snippet |
|---|---|---|
| `plugin.py:29` | `api.logger` | `log = api.logger("svrnty.plugin")` |
| `plugin.py:34` | `api.register_static` | `api.register_static(STATIC_PREFIX, str(STATIC_DIR))` |
| `plugin.py:35` | `api.inject_stylesheet` | `api.inject_stylesheet(f"/plugins/{STATIC_PREFIX}/app.css")` |
| `plugin.py:36` | `api.inject_script` | `api.inject_script(f"/plugins/{STATIC_PREFIX}/app.js")` |
| `plugin.py:39` | `api.inject_script` | `api.inject_script(f"/plugins/{STATIC_PREFIX}/svrnty_nav.js")` |
| `plugin.py:42` | `api.inject_stylesheet` | `api.inject_stylesheet(f"/plugins/{STATIC_PREFIX}/adwright.css")` |
| `plugin.py:43` | `api.inject_script` | `api.inject_script(f"/plugins/{STATIC_PREFIX}/adwright.js")` |
| `plugin.py:45` | `api.inject_stylesheet` | `api.inject_stylesheet(f"/plugins/{STATIC_PREFIX}/bte.css")` |
| `plugin.py:46` | `api.inject_script` | `api.inject_script(f"/plugins/{STATIC_PREFIX}/bte.js")` |
| `plugin.py:48` | `api.inject_stylesheet` | `api.inject_stylesheet(f"/plugins/{STATIC_PREFIX}/umbrella_inline.css")` |
| `plugin.py:49` | `api.inject_script` | `api.inject_script(f"/plugins/{STATIC_PREFIX}/umbrella_inline.js")` |
| `plugin.py:31` | `api.logger` | `log = api.logger("svrnty.plugin")` |
| `plugin.py:36` | `api.register_static` | `api.register_static(STATIC_PREFIX, str(STATIC_DIR))` |
| `plugin.py:37` | `api.inject_stylesheet` | `api.inject_stylesheet(f"/plugins/{STATIC_PREFIX}/app.css")` |
| `plugin.py:38` | `api.inject_script` | `api.inject_script(f"/plugins/{STATIC_PREFIX}/app.js")` |
| `plugin.py:41` | `api.inject_script` | `api.inject_script(f"/plugins/{STATIC_PREFIX}/svrnty_nav.js")` |
| `plugin.py:44` | `api.inject_stylesheet` | `api.inject_stylesheet(f"/plugins/{STATIC_PREFIX}/adwright.css")` |
| `plugin.py:45` | `api.inject_script` | `api.inject_script(f"/plugins/{STATIC_PREFIX}/adwright.js")` |
| `plugin.py:47` | `api.inject_stylesheet` | `api.inject_stylesheet(f"/plugins/{STATIC_PREFIX}/bte.css")` |
| `plugin.py:48` | `api.inject_script` | `api.inject_script(f"/plugins/{STATIC_PREFIX}/bte.js")` |
| `plugin.py:50` | `api.inject_stylesheet` | `api.inject_stylesheet(f"/plugins/{STATIC_PREFIX}/canvas.css")` |
| `plugin.py:51` | `api.inject_script` | `api.inject_script(f"/plugins/{STATIC_PREFIX}/canvas.js")` |
| `plugin.py:53` | `api.inject_stylesheet` | `api.inject_stylesheet(f"/plugins/{STATIC_PREFIX}/umbrella_inline.css")` |
| `plugin.py:54` | `api.inject_script` | `api.inject_script(f"/plugins/{STATIC_PREFIX}/umbrella_inline.js")` |
| `routes/adwright.py:68` | `api.logger` | `log = api.logger("svrnty.routes.adwright")` |
| `routes/adwright.py:69` | `api.register_route` | `api.register_route(` |
| `routes/adwright.py:71` | `api.register_route` | `api.register_route(` |
| `routes/bte_proxy.py:90` | `api.logger` | `log = api.logger("svrnty.routes.bte_proxy")` |
| `routes/bte_proxy.py:91` | `api.register_route` | `api.register_route("/api/bte/proxy", "GET", _handle_proxy)` |
| `routes/bte_proxy.py:92` | `api.register_route` | `api.register_route("/api/bte/proxy", "POST", _handle_proxy)` |
| `routes/canvas.py:51` | `api.logger` | `log = api.logger("svrnty.routes.canvas")` |
| `routes/canvas.py:52` | `api.register_route` | `api.register_route("/api/canvas/status", "GET", _handle_status)` |
| `routes/canvas.py:53` | `api.register_route` | `api.register_route("/api/canvas/tools", "GET", _handle_tools)` |
| `routes/canvas.py:54` | `api.register_route` | `api.register_route("/api/canvas/proxy", "GET", _handle_proxy)` |
| `routes/canvas.py:55` | `api.register_route` | `api.register_route("/api/canvas/proxy", "POST", _handle_proxy)` |
| `routes/canvas.py:56` | `api.register_route` | `api.register_route("/api/canvas/proxy", "PUT", _handle_proxy)` |
| `routes/canvas.py:57` | `api.register_route` | `api.register_route("/api/canvas/command", "POST", _handle_command)` |
| `routes/canvas.py:58` | `api.register_route` | `api.register_route("/api/canvas/design-context", "GET", _handle_design_context)` |
| `routes/canvas.py:59` | `api.register_route` | `api.register_route("/api/canvas/events", "GET", _handle_events)` |
| `routes/cortex_os_extension.py:17` | `api.logger` | `log = api.logger("svrnty.routes.cortex_os_extension")` |
| `routes/cortex_os_extension.py:21` | `api.inject_stylesheet` | `api.inject_stylesheet(stylesheet)` |
| `routes/cortex_os_extension.py:22` | `api.inject_script` | `api.inject_script(script)` |
| `routes/cortex_os_runtime_health.py:30` | `api.logger` | `log = api.logger("svrnty.routes.cortex_os_runtime_health")` |
| `routes/cortex_os_runtime_health.py:31` | `api.register_route` | `api.register_route(ROUTE_PATH, ROUTE_METHOD, _handle_runtime_health)` |
| `routes/transcribe.py:37` | `api.logger` | `log = api.logger("svrnty.routes.transcribe")` |
| `routes/transcribe.py:38` | `api.register_route` | `api.register_route("/api/transcribe", "POST", _handle_transcribe)` |
| `routes/transcribe.py:39` | `api.register_audio_attachment_processor` | `api.register_audio_attachment_processor(_transcribe_audio_attachments)` |
@ -54,6 +70,17 @@ _None. Plugin uses only the public API._ ✓
| File | Line | URL |
|---|---|---|
| `static/canvas.js` | 136 | `/api/canvas/status` |
| `static/canvas.js` | 137 | `/api/canvas/tools` |
| `static/canvas.js` | 162 | `/api/canvas/events` |
| `static/canvas.js` | 252 | `/api/canvas/command` |
| `static/canvas.js` | 307 | `/api/v1/projects` |
| `static/canvas.js` | 310 | `/api/v1/projects` |
| `static/canvas.js` | 323 | `/api/v1/projects/` |
| `static/canvas.js` | 339 | `/api/v1/projects/` |
| `static/canvas.js` | 476 | `/api/v1/projects/` |
| `static/canvas.js` | 516 | `/api/v1/projects/` |
| `static/canvas.js` | 542 | `/api/v1/projects/` |
| `static/bte.js` | 365 | `/api/command/requestPhotoshoot` |
| `static/bte.js` | 396 | `/api/query/assetDtos` |
| `static/bte.js` | 408 | `/api/assets/` |
@ -63,4 +90,5 @@ _None. Plugin uses only the public API._ ✓
| `static/adwright.js` | 606 | `/api/adwright/provision-creds` |
| `static/umbrella.js` | 57 | `/api/umbrella` |
| `static/app.js` | 165 | `/api/vault/status` |
| `static/cortex-os/runtime-health/runtime_health.js` | 4 | `/api/cortex-os/runtime-health` |

View File

@ -33,12 +33,16 @@ assets:
- /plugins/svrnty/svrnty_nav.js
- /plugins/svrnty/adwright.js
- /plugins/svrnty/bte.js
- /plugins/svrnty/canvas.js
- /plugins/svrnty/umbrella_inline.js
- /plugins/svrnty/cortex-os/runtime-health/runtime_health.js
stylesheets:
- /plugins/svrnty/app.css
- /plugins/svrnty/adwright.css
- /plugins/svrnty/bte.css
- /plugins/svrnty/canvas.css
- /plugins/svrnty/umbrella_inline.css
- /plugins/svrnty/cortex-os/runtime-health/runtime_health.css
# Routes this plugin registers at load time (declarative cross-check vs runtime).
# Each row maps to a routes/<file>.py.
@ -49,6 +53,15 @@ routes:
- { path: /api/adwright/provision-creds, method: POST, file: routes/adwright.py, status: live }
- { path: /api/bte/proxy, method: GET, file: routes/bte_proxy.py, status: live }
- { path: /api/bte/proxy, method: POST, file: routes/bte_proxy.py, status: live }
- { path: /api/canvas/status, method: GET, file: routes/canvas.py, status: live }
- { path: /api/canvas/tools, method: GET, file: routes/canvas.py, status: live }
- { path: /api/canvas/proxy, method: GET, file: routes/canvas.py, status: live }
- { path: /api/canvas/proxy, method: POST, file: routes/canvas.py, status: live }
- { path: /api/canvas/proxy, method: PUT, file: routes/canvas.py, status: live }
- { path: /api/canvas/command, method: POST, file: routes/canvas.py, status: live }
- { path: /api/canvas/design-context, method: GET, file: routes/canvas.py, status: seed }
- { path: /api/canvas/events, method: GET, file: routes/canvas.py, status: seed }
- { path: /api/cortex-os/runtime-health, method: GET, file: routes/cortex_os_extension.py, status: live }
# Audio-attachment processors (called by streaming.py before agent receives message).
audio_processors:

View File

@ -14,6 +14,8 @@ in `static/`. The map of every upstream dependency is in CONNECTION-MAP.md
import os
from pathlib import Path
from routes import cortex_os_extension
# Static + asset URL prefix (per protocol §12, decision Q5: /plugins/svrnty/<asset>)
STATIC_PREFIX = "svrnty"
STATIC_DIR = Path(__file__).resolve().parent / "static"
@ -44,11 +46,17 @@ def register(api):
# BTE Command Center panel — same pattern (main.svrnty-showing-bte).
api.inject_stylesheet(f"/plugins/{STATIC_PREFIX}/bte.css")
api.inject_script(f"/plugins/{STATIC_PREFIX}/bte.js")
# Sovereign Canvas panel — Stitch-like live design command surface.
api.inject_stylesheet(f"/plugins/{STATIC_PREFIX}/canvas.css")
api.inject_script(f"/plugins/{STATIC_PREFIX}/canvas.js")
# Inline Umbrella graph for the Hermes Workspace right panel.
api.inject_stylesheet(f"/plugins/{STATIC_PREFIX}/umbrella_inline.css")
api.inject_script(f"/plugins/{STATIC_PREFIX}/umbrella_inline.js")
log.info("static + assets wired at /plugins/%s/", STATIC_PREFIX)
# Cortex OS Plugin development mount: S23 Runtime Health read-only member.
cortex_os_extension.register(api, STATIC_PREFIX)
# Routes — each feature lives in its own module under routes/.
# Phase 2 will populate these. Import-and-register pattern; failures are
# logged but don't take down the rest of the plugin.
@ -76,5 +84,6 @@ def _phase2_routes():
"vault_status", # P2.B — vault connections status ✓
"adwright", # P2.C — Adwright tool panel routes (PRD §5+§6) ✓
"bte_proxy", # P2.D — BTE Command Center same-origin proxy (PRD §3) ✓
"canvas", # Sovereign Stitch-like design canvas proxy + event seed
"umbrella", # P2.E — cortex-os umbrella graph viz (UMBRELLA-VIZ-PRD) ✓
]

796
routes/canvas.py Normal file
View File

@ -0,0 +1,796 @@
"""Canvas command-center routes.
These routes are intentionally small: Hermes WebUI owns the command surface,
while the existing canva-editor Go service owns canonical project/screen state.
The proxy keeps browser calls same-origin and whitelisted so the first Stitch
slice can move without coupling WebUI to the design service internals.
"""
import json
import os
import re
import sqlite3
import subprocess
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
_DEFAULT_CANVA_BASE = "http://localhost:8080"
_PLANB_BRAND_ID = "607740e8-8b76-4455-9c4f-eadbe9b4168c"
_BRAND_ALIASES = {
"planb": _PLANB_BRAND_ID,
"plan-b": _PLANB_BRAND_ID,
"goutez-planb": _PLANB_BRAND_ID,
}
_DEFAULT_CMO_DB_PATHS = (
"~/.hermes/profiles/cmo-planb/cmo.db",
"~/.hermes/cmo-planb/cmo.db",
"~/.hermes/cmo/cmo.db",
)
_ALLOWED_EXACT = frozenset({
"/health",
"/api/v1/capabilities",
"/api/v1/projects",
"/api/v1/generate",
"/api/v1/generate/from-image",
})
_ALLOWED_PATTERN = re.compile(
r"^/api/v1/projects/[A-Za-z0-9_\-]+/(screens|design-system|variants|prototype-edges|export)"
r"(/[A-Za-z0-9_\-]+)?$"
)
_EVENT_LOCK = threading.Lock()
_EVENTS = []
_EVENT_CURSOR = 0
_MAX_EVENTS = 200
def register(api):
"""Wire the Canvas proxy, health, design-context, and SSE seed routes."""
log = api.logger("svrnty.routes.canvas")
api.register_route("/api/canvas/status", "GET", _handle_status)
api.register_route("/api/canvas/tools", "GET", _handle_tools)
api.register_route("/api/canvas/proxy", "GET", _handle_proxy)
api.register_route("/api/canvas/proxy", "POST", _handle_proxy)
api.register_route("/api/canvas/proxy", "PUT", _handle_proxy)
api.register_route("/api/canvas/command", "POST", _handle_command)
api.register_route("/api/canvas/design-context", "GET", _handle_design_context)
api.register_route("/api/canvas/events", "GET", _handle_events)
log.info("canvas endpoints registered")
def _handle_status(handler, _parsed):
base = _base_url()
ok = False
detail = "unreachable"
capabilities = None
try:
req = urllib.request.Request(base + "/health", method="GET")
with urllib.request.urlopen(req, timeout=2) as resp:
ok = 200 <= resp.status < 300
detail = "ok" if ok else f"status {resp.status}"
if ok:
caps_req = urllib.request.Request(base + "/api/v1/capabilities", method="GET")
with urllib.request.urlopen(caps_req, timeout=2) as caps_resp:
if 200 <= caps_resp.status < 300:
capabilities = json.loads(caps_resp.read().decode("utf-8"))
except Exception as e:
detail = str(e)
return _send_json(handler, {
"ok": ok,
"service": "canva-editor",
"base_url": base,
"detail": detail,
"capabilities": capabilities,
"boundary": "Hermes WebUI plugin -> canva-editor HTTP API",
"entrypoint": "/plugins/svrnty/canvas.html",
}, 200)
def _handle_tools(handler, _parsed):
return _send_json(handler, {
"ok": True,
"tool_surface": "cmo-canvas-v0",
"tools": [
{
"name": "canvas.status",
"method": "GET",
"path": "/api/canvas/status",
"purpose": "Check canva-editor reachability from Hermes WebUI.",
},
{
"name": "canvas.create_project",
"method": "POST",
"path": "/api/canvas/proxy?path=/api/v1/projects",
"purpose": "Create an owned design project.",
},
{
"name": "canvas.generate_screen",
"method": "POST",
"path": "/api/canvas/command",
"purpose": "Ask the Canvas bridge to generate and persist a screen with live progress events.",
},
{
"name": "canvas.command",
"method": "POST",
"path": "/api/canvas/command",
"purpose": "CMO-facing command endpoint for prompt-to-canvas generation, variants, and event replay.",
},
{
"name": "canvas.generate_from_image",
"method": "POST",
"path": "/api/canvas/proxy?path=/api/v1/generate/from-image",
"purpose": "Generate a structured screen spec from image input.",
},
{
"name": "canvas.save_screen",
"method": "POST",
"path": "/api/canvas/proxy?path=/api/v1/projects/{project_id}/screens",
"purpose": "Persist generated screen specs as canonical canva-editor state.",
},
{
"name": "canvas.update_screen",
"method": "POST",
"path": "/api/canvas/proxy?path=/api/v1/projects/{project_id}/screens/{screen_id}&method=PUT",
"purpose": "Persist simple structured edits to an existing screen spec.",
},
{
"name": "canvas.design_context",
"method": "GET",
"path": "/api/canvas/design-context",
"purpose": "Seed brand and memory context for CMO canvas work.",
},
{
"name": "canvas.record_variant",
"method": "POST",
"path": "/api/canvas/proxy?path=/api/v1/projects/{id}/variants",
"purpose": "Record parent/child lineage for generated design variations.",
},
{
"name": "canvas.connect_prototype_edge",
"method": "POST",
"path": "/api/canvas/proxy?path=/api/v1/projects/{id}/prototype-edges",
"purpose": "Record a navigable relationship between two screens.",
},
{
"name": "canvas.export_project",
"method": "GET",
"path": "/api/canvas/proxy?path=/api/v1/projects/{id}/export",
"purpose": "Export project JSON with screens, variants, prototype edges, and design system.",
},
{
"name": "canvas.events",
"method": "GET",
"path": "/api/canvas/events",
"purpose": "SSE contract seed for live canvas updates.",
},
],
"non_goals": [
"full-canva-parity",
"full-figma-parity",
"publishing-without-jp-approval",
"upstream-hermes-webui-edits",
],
}, 200)
def _handle_design_context(handler, parsed):
qs = urllib.parse.parse_qs(parsed.query or "")
brand_id = (qs.get("brand_id") or ["planb"])[0].strip() or "planb"
context = _load_design_context(brand_id)
return _send_json(handler, context, 200)
def _handle_events(handler, _parsed):
"""Return replayable Canvas events.
The canva-editor service does not expose a live event bus yet. This route
establishes the browser/backend contract now. The Canvas panel polls JSON
deltas, while SSE callers still receive a valid seed stream plus replay.
"""
qs = urllib.parse.parse_qs(getattr(_parsed, "query", "") or "")
if "since" in qs or (qs.get("format") or [""])[0] == "json":
since = _safe_int((qs.get("since") or ["0"])[0], 0)
events = _events_since(since)
cursor = events[-1]["id"] if events else _current_event_cursor()
return _send_json(handler, {"ok": True, "cursor": cursor, "events": events}, 200)
replay = _events_since(0)[-20:]
chunks = [
"event: canvas.connected\n"
"data: {\"ok\":true,\"source\":\"svrnty-hermes-webui-plugin\"}\n\n"
]
for event in replay:
event_name = event.get("type", "canvas.event")
chunks.append(f"event: {event_name}\n")
chunks.append("data: " + json.dumps(event) + "\n\n")
body = "".join(chunks).encode("utf-8")
handler.send_response(200)
handler.send_header("Content-Type", "text/event-stream; charset=utf-8")
handler.send_header("Cache-Control", "no-store")
handler.send_header("Connection", "close")
handler.send_header("Content-Length", str(len(body)))
handler.end_headers()
handler.wfile.write(body)
return True
def _handle_command(handler, _parsed):
"""Run a CMO-style prompt-to-canvas command through canva-editor."""
try:
body = _read_json_body(handler)
except ValueError as e:
return _send_json(handler, {"ok": False, "error": str(e)}, 400)
prompt = str(body.get("prompt") or "").strip()
if not prompt:
return _send_json(handler, {"ok": False, "error": "prompt is required"}, 400)
variants = max(1, min(6, _safe_int(body.get("variants"), 1)))
brand_id = str(body.get("brand_id") or body.get("brandId") or "planb").strip() or "planb"
project_name = str(body.get("project_name") or body.get("projectName") or "Hermes Canvas").strip() or "Hermes Canvas"
parent_screen_id = str(body.get("parent_screen_id") or body.get("parentScreenId") or "").strip()
command_id = f"canvas-{int(time.time() * 1000)}"
first_screen_id = parent_screen_id
screens = []
emitted_start = _current_event_cursor()
design_context = _load_design_context(brand_id)
_emit_event("canvas.command.accepted", {
"command_id": command_id,
"prompt": prompt,
"variants": variants,
"brand_id": brand_id,
"design_context_status": design_context.get("status"),
"design_context_version": design_context.get("source_version"),
})
try:
project = _ensure_project(project_name)
project_id = _record_id(project)
if not project_id:
raise RuntimeError("canva-editor project id missing")
_emit_event("canvas.project.ready", {
"command_id": command_id,
"project": project,
})
for idx in range(variants):
user_variant_prompt = prompt if variants == 1 else (
f"{prompt}\nVariant {idx + 1} of {variants}: choose a distinct composition."
)
variant_prompt = _compose_generation_prompt(user_variant_prompt, design_context)
_emit_event("canvas.variant.started", {
"command_id": command_id,
"project_id": project_id,
"variant_index": idx + 1,
"variants": variants,
"design_context_status": design_context.get("status"),
})
generated = _canva_json("/api/v1/generate", "POST", {"prompt": variant_prompt}, timeout=90)
spec = generated.get("screenSpec") or generated.get("screen_spec") or generated
screen = _canva_json(
f"/api/v1/projects/{urllib.parse.quote(project_id)}/screens",
"POST",
{"name": prompt[:42] or "Generated screen", "spec": spec},
timeout=30,
)
screen_id = _record_id(screen)
if idx == 0 and not first_screen_id:
first_screen_id = screen_id
elif first_screen_id and screen_id:
_canva_json(
f"/api/v1/projects/{urllib.parse.quote(project_id)}/variants",
"POST",
{
"parentScreenId": first_screen_id,
"screenId": screen_id,
"prompt": user_variant_prompt,
"aspects": ["layout", "copy", "composition"],
"metadata": {"commandId": command_id, "variantIndex": idx + 1},
},
timeout=30,
)
item = {
"screen": screen,
"screenSpec": spec,
"variant_index": idx + 1,
"prompt": user_variant_prompt,
"designContext": _compact_design_context(design_context),
}
screens.append(item)
_emit_event("canvas.screen.persisted", {
"command_id": command_id,
"project_id": project_id,
**item,
})
_emit_event("canvas.command.completed", {
"command_id": command_id,
"project": project,
"screens": len(screens),
})
emitted = _events_since(emitted_start)
return _send_json(handler, {
"ok": True,
"command_id": command_id,
"project": project,
"screens": screens,
"design_context": _compact_design_context(design_context),
"events": emitted,
"cursor": emitted[-1]["id"] if emitted else _current_event_cursor(),
}, 200)
except urllib.error.HTTPError as e:
detail = e.read().decode("utf-8", errors="replace") if e.fp else str(e)
_emit_event("canvas.command.failed", {"command_id": command_id, "error": detail})
return _send_json(handler, {"ok": False, "command_id": command_id, "error": detail}, e.code)
except urllib.error.URLError as e:
detail = f"canva-editor unreachable: {e.reason}"
_emit_event("canvas.command.failed", {"command_id": command_id, "error": detail})
return _send_json(handler, {"ok": False, "command_id": command_id, "error": detail}, 502)
except Exception as e:
detail = str(e)
_emit_event("canvas.command.failed", {"command_id": command_id, "error": detail})
return _send_json(handler, {"ok": False, "command_id": command_id, "error": detail}, 500)
def _handle_proxy(handler, parsed):
qs = urllib.parse.parse_qs(parsed.query or "")
target_path = (qs.get("path") or [""])[0].strip()
method_override = (qs.get("method") or [""])[0].strip().upper()
if not target_path.startswith(("/health", "/api/v1/")):
return _send_json(handler, {"ok": False, "error": "missing or invalid path"}, 400)
if not _is_allowed(target_path):
return _send_json(handler, {"ok": False, "error": f"path not allowed: {target_path}"}, 403)
method = handler.command
if method == "POST" and method_override in {"PUT"}:
method = method_override
body = b""
if method in ("POST", "PUT", "PATCH"):
length = int(handler.headers.get("Content-Length", "0") or 0)
if length > 0:
body = handler.rfile.read(length)
target_url = _base_url() + target_path
req = urllib.request.Request(target_url, data=body if body else None, method=method)
if body:
req.add_header("Content-Type", handler.headers.get("Content-Type", "application/json"))
try:
with urllib.request.urlopen(req, timeout=45) as resp:
resp_body = resp.read()
resp_ctype = resp.headers.get("Content-Type", "application/json")
resp_status = resp.status
except urllib.error.HTTPError as e:
resp_body = e.read() or b""
resp_ctype = e.headers.get("Content-Type", "application/json") if e.headers else "application/json"
resp_status = e.code
except urllib.error.URLError as e:
return _send_json(handler, {"ok": False, "error": f"canva-editor unreachable: {e.reason}"}, 502)
except Exception as e:
return _send_json(handler, {"ok": False, "error": f"proxy error: {e}"}, 500)
handler.send_response(resp_status)
handler.send_header("Content-Type", resp_ctype)
handler.send_header("Content-Length", str(len(resp_body)))
handler.send_header("Cache-Control", "no-store")
handler.end_headers()
handler.wfile.write(resp_body)
return True
def _is_allowed(path: str) -> bool:
return path in _ALLOWED_EXACT or bool(_ALLOWED_PATTERN.match(path))
def _base_url() -> str:
return os.environ.get("CANVA_EDITOR_BASE_URL", _DEFAULT_CANVA_BASE).rstrip("/")
def _load_design_context(brand_id: str) -> dict:
resolved_brand_id = _resolve_brand_id(brand_id)
base_context = {
"brand_id": brand_id,
"resolved_brand_id": resolved_brand_id,
"source": "cmo.brand_profile_cache",
"status": "missing",
"brand": None,
"brand_guideline_text": "",
"design_md": "",
"palette_tokens": [],
"typography_tokens": [],
"spacing_tokens": [],
"voice_rules": [],
"memory_hints": [],
"secondbrain": {"status": "not_checked", "hints": []},
"deferred_ports": ["secondbrain-writeback", "svrnty-vision"],
}
db_path = _cmo_db_path()
base_context["cmo_db_path"] = db_path
if not db_path:
base_context["error"] = "CMO database not found"
return base_context
try:
row = _read_brand_cache(db_path, resolved_brand_id)
except sqlite3.Error as e:
base_context["status"] = "error"
base_context["error"] = f"CMO database read failed: {e}"
return base_context
if not row:
base_context["error"] = f"brand_profile_cache row not found for {resolved_brand_id}"
return base_context
try:
blob = json.loads(row["json"] or "{}")
except ValueError as e:
base_context["status"] = "error"
base_context["error"] = f"brand_profile_cache JSON invalid: {e}"
return base_context
brand = blob.get("brand") if isinstance(blob.get("brand"), dict) else None
guideline = str(blob.get("brand_guideline") or "")
design_md = str(blob.get("design_md") or "")
palette = _normalise_token_list(blob.get("palette"))
typography = _normalise_token_list(blob.get("typography"))
spacing = _normalise_token_list(blob.get("spacing"))
voice_rules = _extract_voice_rules(blob.get("voice"), guideline)
memory_hints = _build_memory_hints(brand, guideline, palette, typography, spacing, voice_rules)
secondbrain = _load_secondbrain_hints(brand, resolved_brand_id)
memory_hints.extend(secondbrain.get("hints") or [])
base_context.update({
"status": "ready",
"cache_brand_id": row["brand_id"],
"source_version": row["version"],
"fetched_at": row["fetched_at"],
"brand": brand,
"brand_guideline_text": guideline,
"brand_guideline_excerpt": _excerpt(guideline, 1800),
"design_md": design_md,
"design_md_excerpt": _excerpt(design_md, 1400),
"palette_tokens": palette,
"typography_tokens": typography,
"spacing_tokens": spacing,
"voice_rules": voice_rules,
"memory_hints": memory_hints[:14],
"secondbrain": secondbrain,
"tokens_imported": bool(blob.get("tokens_imported") or palette or typography or spacing),
})
return base_context
def _resolve_brand_id(brand_id: str) -> str:
value = (brand_id or "").strip()
return _BRAND_ALIASES.get(value.lower(), value)
def _cmo_db_path() -> str:
candidates = []
for env_name in ("CANVAS_CMO_DB", "CMO_DB"):
env_path = os.environ.get(env_name)
if env_path:
candidates.append(env_path)
candidates.extend(_DEFAULT_CMO_DB_PATHS)
for path in candidates:
expanded = os.path.expanduser(path)
if os.path.exists(expanded):
return expanded
return ""
def _read_brand_cache(db_path: str, brand_id: str):
conn = sqlite3.connect(db_path)
conn.row_factory = sqlite3.Row
try:
row = conn.execute(
"""SELECT brand_id, json, version, fetched_at
FROM brand_profile_cache
WHERE brand_id = ?""",
(brand_id,),
).fetchone()
return row
finally:
conn.close()
def _normalise_token_list(raw_tokens) -> list:
if not isinstance(raw_tokens, list):
return []
tokens = []
for item in raw_tokens[:80]:
if not isinstance(item, dict):
continue
token = {
"path": item.get("path") or item.get("name") or item.get("key"),
"type": item.get("type"),
"description": item.get("description"),
}
value = item.get("value")
if value is None:
value = item.get("valueJson")
if isinstance(value, str):
try:
value = json.loads(value)
except ValueError:
pass
token["value"] = value
tokens.append({k: v for k, v in token.items() if v not in (None, "")})
return tokens
def _extract_voice_rules(voice, guideline: str) -> list:
rules = []
if isinstance(voice, dict):
for key in ("tone", "voice", "dos", "donts", "rules", "pillars", "description"):
value = voice.get(key) or voice.get(key[0].upper() + key[1:])
if isinstance(value, str) and value.strip():
rules.append(value.strip())
elif isinstance(value, list):
rules.extend(str(item).strip() for item in value if str(item).strip())
elif isinstance(voice, str) and voice.strip():
rules.append(voice.strip())
if not rules and guideline:
for line in guideline.splitlines():
clean = line.strip(" -\t")
if clean and any(word in clean.lower() for word in ("voice", "tone", "avoid", "must", "never")):
rules.append(clean)
if len(rules) >= 8:
break
return rules[:12]
def _build_memory_hints(brand, guideline, palette, typography, spacing, voice_rules) -> list:
hints = []
if isinstance(brand, dict):
name = brand.get("displayName") or brand.get("name")
description = brand.get("description")
if name:
hints.append(f"Brand: {name}")
if description:
hints.append(f"Brand description: {_excerpt(str(description), 260)}")
if palette:
colors = []
for token in palette:
value = token.get("value")
if isinstance(value, dict):
colors.extend(str(v) for v in value.values() if str(v).startswith("#"))
elif isinstance(value, str) and value.startswith("#"):
colors.append(value)
if colors:
hints.append("Use BTE palette tokens: " + ", ".join(dict.fromkeys(colors[:8])))
if typography:
type_paths = [str(t.get("path")) for t in typography if t.get("path")]
if type_paths:
hints.append("Typography context: " + ", ".join(type_paths[:8]))
if spacing:
hints.append("Respect BTE spacing tokens and density rhythm.")
if voice_rules:
hints.append("Voice constraints: " + " | ".join(_excerpt(rule, 180) for rule in voice_rules[:3]))
if guideline:
hints.append("Use BTE brand guideline as primary public-plane creative truth.")
return hints[:10]
def _load_secondbrain_hints(brand, brand_id: str) -> dict:
"""Read a few relevant knowledge capsules through local psql when configured."""
password = os.environ.get("SECONDBRAIN_DB_PASSWORD") or os.environ.get("PGPASSWORD")
database_url = os.environ.get("SECONDBRAIN_DATABASE_URL")
if not database_url and not password:
return {"status": "unconfigured", "hints": []}
psql = os.environ.get("PSQL_BIN", "psql")
sql = """
SELECT COALESCE(json_agg(row_to_json(t))::text, '[]')
FROM (
SELECT id, title, summary, sector, created_at
FROM knowledge_capsules
WHERE sector IN ('svrnty', 'planb', 'personal')
AND (
title ILIKE '%design%' OR summary ILIKE '%design%' OR content ILIKE '%design%' OR
title ILIKE '%brand%' OR summary ILIKE '%brand%' OR content ILIKE '%brand%' OR
title ILIKE '%canvas%' OR summary ILIKE '%canvas%' OR content ILIKE '%canvas%' OR
title ILIKE '%stitch%' OR summary ILIKE '%stitch%' OR content ILIKE '%stitch%'
)
ORDER BY created_at DESC
LIMIT 4
) AS t;
""".strip()
cmd = [psql]
env = os.environ.copy()
if database_url:
cmd.append(database_url)
else:
env["PGPASSWORD"] = password
cmd.extend([
"-h", os.environ.get("SECONDBRAIN_DB_HOST", "localhost"),
"-p", os.environ.get("SECONDBRAIN_DB_PORT", "5435"),
"-U", os.environ.get("SECONDBRAIN_DB_USER", "svrnty"),
"-d", os.environ.get("SECONDBRAIN_DB_NAME", "secondbrain"),
])
cmd.extend(["-tA", "-c", sql])
try:
result = subprocess.run(
cmd,
env=env,
check=False,
capture_output=True,
text=True,
timeout=2.5,
)
except FileNotFoundError:
return {"status": "psql_missing", "hints": []}
except subprocess.TimeoutExpired:
return {"status": "timeout", "hints": []}
except Exception as e:
return {"status": "error", "hints": [], "error": str(e)}
if result.returncode != 0:
return {"status": "error", "hints": [], "error": _excerpt(result.stderr, 240)}
try:
rows = json.loads((result.stdout or "[]").strip() or "[]")
except ValueError as e:
return {"status": "error", "hints": [], "error": f"invalid secondbrain JSON: {e}"}
if not isinstance(rows, list):
rows = []
hints = []
capsules = []
brand_name = ""
if isinstance(brand, dict):
brand_name = brand.get("displayName") or brand.get("name") or ""
for row in rows[:4]:
if not isinstance(row, dict):
continue
title = str(row.get("title") or "Untitled capsule")
summary = _excerpt(str(row.get("summary") or ""), 260)
capsule = {
"id": row.get("id"),
"title": title,
"sector": row.get("sector"),
"created_at": row.get("created_at"),
}
capsules.append({k: v for k, v in capsule.items() if v not in (None, "")})
hint = f"Secondbrain capsule {row.get('id')}: {title}"
if summary:
hint += f"{summary}"
hints.append(hint)
return {
"status": "ready",
"source": "secondbrain.knowledge_capsules",
"brand_id": brand_id,
"brand_name": brand_name,
"capsules": capsules,
"hints": hints,
}
def _excerpt(text: str, limit: int) -> str:
cleaned = " ".join((text or "").split())
if len(cleaned) <= limit:
return cleaned
return cleaned[: max(0, limit - 3)].rstrip() + "..."
def _compact_design_context(context: dict) -> dict:
return {
"brand_id": context.get("brand_id"),
"resolved_brand_id": context.get("resolved_brand_id"),
"status": context.get("status"),
"source": context.get("source"),
"source_version": context.get("source_version"),
"fetched_at": context.get("fetched_at"),
"tokens_imported": context.get("tokens_imported"),
"memory_hints": context.get("memory_hints") or [],
"secondbrain_status": (context.get("secondbrain") or {}).get("status"),
}
def _compose_generation_prompt(prompt: str, context: dict) -> str:
if not context or context.get("status") != "ready":
return prompt
pieces = [prompt.strip()]
brand = context.get("brand") or {}
brand_name = brand.get("displayName") or brand.get("name")
if brand_name:
pieces.append(f"Brand context: {brand_name}.")
hints = context.get("memory_hints") or []
if hints:
pieces.append("BTE/CMO design constraints:\n- " + "\n- ".join(str(h) for h in hints[:8]))
guideline = context.get("brand_guideline_excerpt") or context.get("design_md_excerpt") or ""
if guideline:
pieces.append("Brand guideline excerpt:\n" + _excerpt(guideline, 1600))
pieces.append("Generate draft-only canvas output for Hermes WebUI; do not publish or imply approval.")
return "\n\n".join(piece for piece in pieces if piece)
def _read_json_body(handler) -> dict:
length = int(handler.headers.get("Content-Length", "0") or 0)
if length <= 0:
return {}
raw = handler.rfile.read(length)
try:
payload = json.loads(raw.decode("utf-8"))
except Exception as e:
raise ValueError(f"invalid JSON: {e}")
if not isinstance(payload, dict):
raise ValueError("body must be a JSON object")
return payload
def _ensure_project(name: str) -> dict:
projects = _canva_json("/api/v1/projects", "GET", None, timeout=15)
if isinstance(projects, list) and projects:
return projects[0]
return _canva_json("/api/v1/projects", "POST", {"name": name}, timeout=15)
def _canva_json(path: str, method: str = "GET", payload=None, timeout: int = 45):
data = None
if payload is not None:
data = json.dumps(payload).encode("utf-8")
req = urllib.request.Request(_base_url() + path, data=data, method=method)
if data is not None:
req.add_header("Content-Type", "application/json")
with urllib.request.urlopen(req, timeout=timeout) as resp:
raw = resp.read()
if not raw:
return {}
return json.loads(raw.decode("utf-8"))
def _record_id(record) -> str:
if not isinstance(record, dict):
return ""
return str(record.get("id") or record.get("ID") or record.get("uuid") or "")
def _emit_event(event_type: str, payload: dict) -> dict:
global _EVENT_CURSOR
with _EVENT_LOCK:
_EVENT_CURSOR += 1
event = {
"id": _EVENT_CURSOR,
"type": event_type,
"at": time.time(),
**payload,
}
_EVENTS.append(event)
del _EVENTS[:-_MAX_EVENTS]
return dict(event)
def _events_since(cursor: int) -> list:
with _EVENT_LOCK:
return [dict(event) for event in _EVENTS if event["id"] > cursor]
def _current_event_cursor() -> int:
with _EVENT_LOCK:
return _EVENT_CURSOR
def _safe_int(value, fallback: int) -> int:
try:
return int(value)
except Exception:
return fallback
def _send_json(handler, payload: dict, status: int) -> bool:
body = json.dumps(payload).encode("utf-8")
handler.send_response(status)
handler.send_header("Content-Type", "application/json; charset=utf-8")
handler.send_header("Content-Length", str(len(body)))
handler.send_header("Cache-Control", "no-store")
handler.end_headers()
handler.wfile.write(body)
return True

View File

@ -0,0 +1,30 @@
"""Cortex OS Plugin Hermes WebUI Host Adapter extraction surface."""
from routes import cortex_os_runtime_health
PACKAGE_ID = "cortex-os-extension"
MEMBER_ID = "s23-runtime-health-read-only-slice"
ACTIVE_DEVELOPMENT_MOUNT_TARGET = "hermes-webui"
SURFACE_ID = "hermes-webui-host-adapter"
RUNTIME_HEALTH_ROUTE = "/api/cortex-os/runtime-health"
RUNTIME_HEALTH_METHOD = "GET"
RUNTIME_HEALTH_STYLESHEET = "/plugins/{static_prefix}/cortex-os/runtime-health/runtime_health.css"
RUNTIME_HEALTH_SCRIPT = "/plugins/{static_prefix}/cortex-os/runtime-health/runtime_health.js"
def register(api, static_prefix):
"""Register only the Cortex OS Runtime Health member for Hermes WebUI."""
log = api.logger("svrnty.routes.cortex_os_extension")
stylesheet = RUNTIME_HEALTH_STYLESHEET.format(static_prefix=static_prefix)
script = RUNTIME_HEALTH_SCRIPT.format(static_prefix=static_prefix)
api.inject_stylesheet(stylesheet)
api.inject_script(script)
cortex_os_runtime_health.register(api)
log.info(
"cortex-os-extension mounted: %s %s via %s",
RUNTIME_HEALTH_METHOD,
RUNTIME_HEALTH_ROUTE,
SURFACE_ID,
)

View File

@ -0,0 +1,212 @@
"""GET /api/cortex-os/runtime-health - Cortex OS Runtime Health slice.
Public API surface used: api.register_route, api.logger.
No forced internal dependencies. This module does not import Hermes internals.
"""
from __future__ import annotations
import json
import re
from typing import Any
ROUTE_PATH = "/api/cortex-os/runtime-health"
ROUTE_METHOD = "GET"
CONTRACT_ID = "runtime-health/v0.1"
CHECKED_AT = "2026-05-29T00:00:00Z"
SCHEMA_VERSION = "0.1.0"
HOST_RUNTIME_ID = "webui"
HOST_ADAPTER_ID = "hermes"
HOST_ADAPTER_KIND = "development_host"
_FORBIDDEN_TEXT = re.compile(
r"(https?://|/home/|workspaces/|\b\d{2,5}\b|token|secret|cookie|traceback|exception|\.env)",
re.IGNORECASE,
)
def register(api: Any) -> None:
"""Wire the read-only Runtime Health route."""
log = api.logger("svrnty.routes.cortex_os_runtime_health")
api.register_route(ROUTE_PATH, ROUTE_METHOD, _handle_runtime_health)
log.info("cortex os runtime health endpoint registered")
def _handle_runtime_health(handler: Any, parsed: Any) -> bool:
"""Handler signature matches the plugin loader contract."""
if getattr(handler, "command", ROUTE_METHOD) != ROUTE_METHOD:
_write_json(handler, 405, _error_envelope("method_not_allowed", "read only route"))
return True
if getattr(parsed, "query", ""):
_write_json(handler, 400, _error_envelope("query_not_allowed", "query targets are not accepted"))
return True
_write_json(handler, 200, {"ok": True, "result": runtime_health_payload(), "error": None})
return True
def runtime_health_payload(host_signals: dict[str, Any] | None = None) -> dict[str, Any]:
"""Return the host-neutral Runtime Health envelope."""
signals = _summarize_host_signals(host_signals or {})
status = _derive_status(signals)
return {
"schema_version": SCHEMA_VERSION,
"cortex_os_contract_id": CONTRACT_ID,
"host_runtime_id": HOST_RUNTIME_ID,
"host_adapter_id": HOST_ADAPTER_ID,
"host_adapter_kind": HOST_ADAPTER_KIND,
"checked_at": CHECKED_AT,
"status": status,
"readiness": "not_configured",
"signals": signals,
"display_summary": _display_summary_for(status),
"redactions": [
{"target_path": "signals", "reason_code": "no_raw_host_payload", "label": "raw host payloads omitted"},
{"target_path": "signals.summary", "reason_code": "path_redacted", "label": "raw paths redacted"},
{"target_path": "signals.summary", "reason_code": "secret_redacted", "label": "secrets redacted"},
{
"target_path": "source_trace",
"reason_code": "request_response_redacted",
"label": "request and response details omitted",
},
],
"warnings": [
{
"code": "deterministic_host_surface_inputs_only",
"message": "Runtime Health uses deterministic declared host-surface inputs only.",
}
],
"errors": [],
"source_trace": [
{
"source_id": "runtime-health-contract",
"source_kind": "sot_document",
"label": "Runtime Health Contract",
},
{
"source_id": "hermes-host-adapter-contract",
"source_kind": "host_adapter_contract",
"label": "Hermes Host Adapter Contract",
},
{
"source_id": "hermes-runtime-health-slice",
"source_kind": "validator",
"label": "Hermes Runtime Health Slice",
},
],
"authority": {
"runtime_coding": False,
"hermes_source_edits": False,
"hermes_host_adapter_authority_map": False,
"hermes_host_adapter_implementation": False,
"webui_plugin_implementation": False,
"local_json_api_route_files": False,
"local_json_api_route_handlers": False,
"browser_source": False,
"host_runtime_start": False,
"runtime_process_behavior": False,
"runtime_state_mutation": False,
"product_ui_implementation": False,
"display_source": False,
"tool_callable_authority": False,
"mcp_server_runtime_behavior": False,
"mcp_tool_exposure": False,
"profile_exposure_broadening": False,
"memory_domain_access": False,
"delegated_memory_grants": False,
"sharing": False,
"installer_automation": False,
"source_import": False,
"forced_internal_upstream_dependency": False,
"live_smoke_execution": False,
"product_readiness_claim": False,
},
}
def _summarize_host_signals(host_signals: dict[str, Any]) -> list[dict[str, str]]:
names = ["health", "agent_health", "dashboard_status"]
if not host_signals:
return [
{"signal_id": name, "state": "unknown", "summary": "not configured", "redacted": False}
for name in names
]
return [
_summarize_signal(name, host_signals)
for name in names
]
def _summarize_signal(name: str, host_signals: dict[str, Any]) -> dict[str, Any]:
summary, redacted = _bounded_text(host_signals.get(f"{name}_detail", "declared surface"))
return {
"signal_id": name,
"state": _clean_status(host_signals.get(name, "unknown")),
"summary": summary,
"redacted": redacted,
}
def _derive_status(signals: list[dict[str, str]]) -> str:
statuses = {signal["state"] for signal in signals}
if "unavailable" in statuses:
return "unavailable"
if "degraded" in statuses:
return "degraded"
if statuses == {"healthy"}:
return "healthy"
return "unknown"
def _clean_status(value: Any) -> str:
text = str(value).strip().lower()
return text if text in {"healthy", "degraded", "unavailable", "unknown"} else "unknown"
def _display_summary_for(status: str) -> dict[str, str]:
if status == "healthy":
return {
"headline": "Runtime Health signals are healthy.",
"detail": "Declared Runtime Health signals are healthy.",
"severity": "ok",
}
if status == "degraded":
return {
"headline": "Runtime Health signals are degraded.",
"detail": "One or more declared Runtime Health signals are degraded.",
"severity": "warning",
}
if status == "unavailable":
return {
"headline": "Runtime Health signals are unavailable.",
"detail": "One or more declared Runtime Health signals are unavailable.",
"severity": "error",
}
return {
"headline": "Runtime Health is not configured.",
"detail": "Runtime Health has not been live-probed in this slice.",
"severity": "neutral",
}
def _bounded_text(value: Any) -> tuple[str, bool]:
text = str(value).strip().replace("\n", " ")
if not text:
return "redacted", True
if _FORBIDDEN_TEXT.search(text):
return "redacted", True
return text[:160], False
def _error_envelope(code: str, message: str) -> dict[str, Any]:
return {"ok": False, "result": None, "error": {"code": code, "message": message}}
def _write_json(handler: Any, status_code: int, payload: dict[str, Any]) -> None:
body = json.dumps(payload, sort_keys=True).encode("utf-8")
handler.send_response(status_code)
handler.send_header("Content-Type", "application/json; charset=utf-8")
handler.send_header("Content-Length", str(len(body)))
handler.send_header("Cache-Control", "no-store")
handler.end_headers()
handler.wfile.write(body)

View File

@ -28,10 +28,19 @@ PLUGIN_REPO = Path(__file__).resolve().parent.parent
SMOKE = [
{"path": "/health", "expect": [200], "kind": "vanilla"},
{"path": "/api/vault/status", "expect": [200, 401, 403], "kind": "plugin"},
{"path": "/api/canvas/status", "expect": [200, 503, 401, 403], "kind": "plugin"},
{"path": "/api/canvas/tools", "expect": [200, 401, 403], "kind": "plugin"},
{"path": "/api/canvas/proxy?path=/api/v1/capabilities", "expect": [200, 401, 403, 502], "kind": "plugin"},
{"path": "/api/canvas/command", "method": "POST", "body": b"{}", "expect": [400, 401, 403], "kind": "plugin"},
{"path": "/api/canvas/design-context", "expect": [200, 401, 403], "kind": "plugin"},
{"path": "/api/canvas/events", "expect": [200, 401, 403], "kind": "plugin"},
{"path": "/api/umbrella", "expect": [200, 401, 403], "kind": "plugin"},
{"path": "/api/umbrella/doc?path=sot/README.md", "expect": [200, 401, 403], "kind": "plugin"},
{"path": "/plugins/svrnty/app.css", "expect": [200, 302, 401, 403], "kind": "plugin-static"},
{"path": "/plugins/svrnty/app.js", "expect": [200, 302, 401, 403], "kind": "plugin-static"},
{"path": "/plugins/svrnty/canvas.css", "expect": [200, 302, 401, 403], "kind": "plugin-static"},
{"path": "/plugins/svrnty/canvas.js", "expect": [200, 302, 401, 403], "kind": "plugin-static"},
{"path": "/plugins/svrnty/canvas.html", "expect": [200, 302, 401, 403], "kind": "plugin-static"},
{"path": "/plugins/svrnty/umbrella.html", "expect": [200, 302, 401, 403], "kind": "plugin-static"},
{"path": "/plugins/svrnty/umbrella.css", "expect": [200, 302, 401, 403], "kind": "plugin-static"},
{"path": "/plugins/svrnty/umbrella.js", "expect": [200, 302, 401, 403], "kind": "plugin-static"},
@ -61,10 +70,16 @@ def _wait_for(url, timeout=20):
return False
def _hit(base, path):
def _hit(base, spec):
path = spec["path"]
url = base.rstrip("/") + path
method = spec.get("method", "GET")
body = spec.get("body")
try:
with urlopen(url, timeout=5) as r:
req = Request(url, data=body, method=method)
if body is not None:
req.add_header("Content-Type", "application/json")
with urlopen(req, timeout=5) as r:
return r.status, r.read()[:200]
except URLError as e:
if hasattr(e, "code"):
@ -78,7 +93,7 @@ def smoke(base):
rows = []
failed = 0
for s in SMOKE:
status, _body = _hit(base, s["path"])
status, _body = _hit(base, s)
ok = status in s["expect"]
rows.append({"path": s["path"], "status": status, "kind": s["kind"], "ok": ok})
if not ok:

339
scripts/canvas-visual-smoke.py Executable file
View File

@ -0,0 +1,339 @@
#!/usr/bin/env python3
"""Browser smoke for the Hermes Canvas command-center loop.
Starts canva-editor in mock mode, starts hermes-webui with the Svrnty plugin,
opens the real WebUI shell, switches to Canvas, generates variants, captures a
screenshot, and writes a compact JSON proof.
"""
from __future__ import annotations
import argparse
import json
import os
import signal
import socket
import subprocess
import sys
import tempfile
import time
import warnings
from datetime import datetime
from pathlib import Path
from urllib.error import URLError
from urllib.request import urlopen
from PIL import Image
from playwright.sync_api import sync_playwright
PLUGIN_REPO = Path(__file__).resolve().parent.parent
HERMES_REPO = PLUGIN_REPO.parent
WEBUI_REPO = HERMES_REPO / "hermes-webui"
AGENT_REPO = HERMES_REPO / "hermes-agent"
CANVA_REPO = HERMES_REPO.parent / "cortex" / "L2-svrnty.tool-canva-editor"
CMO_REPO = HERMES_REPO / "cmo"
DEFAULT_OUT = HERMES_REPO / ".scratch" / "canvas-visual-proof.json"
def _free_port() -> int:
sock = socket.socket()
sock.bind(("127.0.0.1", 0))
port = sock.getsockname()[1]
sock.close()
return port
def _wait_for(url: str, timeout: int = 30) -> bool:
deadline = time.time() + timeout
while time.time() < deadline:
try:
with urlopen(url, timeout=1) as res:
if res.status < 500:
return True
except URLError:
pass
except Exception:
pass
time.sleep(0.25)
return False
def _pixel_stats(path: Path) -> dict:
img = Image.open(path).convert("RGB")
width, height = img.size
with warnings.catch_warnings():
warnings.simplefilter("ignore", DeprecationWarning)
pixels = list(img.getdata())
total = len(pixels)
bright = sum(1 for r, g, b in pixels if max(r, g, b) > 150)
distinct = len(set(pixels[:: max(1, total // 20000)]))
return {
"path": str(path),
"width": width,
"height": height,
"bright_ratio": round(bright / total, 4),
"sampled_distinct_colors": distinct,
"nonblank": bright > 1000 and distinct > 20,
}
def _start_canva(port: int, db_path: Path) -> subprocess.Popen:
env = os.environ.copy()
env["PATH"] = f"/home/svrnty/sdk/go1.26.0/bin{os.pathsep}{env.get('PATH', '')}"
env["PORT"] = str(port)
env["DB_PATH"] = str(db_path)
return subprocess.Popen(
["go", "run", "./cmd/server", "--mock"],
cwd=CANVA_REPO,
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
preexec_fn=os.setsid,
)
def _start_webui(port: int, state_dir: Path, canva_base: str, plugin_module: str) -> subprocess.Popen:
env = os.environ.copy()
env.pop("HERMES_WEBUI_PASSWORD", None)
env["CANVA_EDITOR_BASE_URL"] = canva_base
env["HERMES_WEBUI_PYTHON_PLUGIN"] = plugin_module
if plugin_module == "svrnty_hermes_webui_refactor_plugin":
env["SVRNTY_WEBUI_DEVELOPMENT_INCLUDE_PROD"] = "1"
env["HERMES_WEBUI_PORT"] = str(port)
env["HERMES_WEBUI_STATE_DIR"] = str(state_dir)
env["HERMES_WEBUI_AGENT_DIR"] = str(AGENT_REPO)
env["HERMES_REPO_ROOT"] = str(HERMES_REPO)
env["PYTHONPATH"] = (
f"{PLUGIN_REPO}{os.pathsep}{HERMES_REPO / 'webui-plugin-development'}{os.pathsep}{AGENT_REPO}"
if not env.get("PYTHONPATH")
else f"{PLUGIN_REPO}{os.pathsep}{HERMES_REPO / 'webui-plugin-development'}{os.pathsep}{AGENT_REPO}{os.pathsep}{env['PYTHONPATH']}"
)
py = WEBUI_REPO / "venv" / "bin" / "python"
agent_py = AGENT_REPO / "venv" / "bin" / "python"
python_exe = str(py) if py.exists() else str(agent_py) if agent_py.exists() else "python3"
return subprocess.Popen(
[python_exe, str(WEBUI_REPO / "server.py")],
cwd=AGENT_REPO if AGENT_REPO.exists() else WEBUI_REPO,
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
preexec_fn=os.setsid,
)
def _stop(proc: subprocess.Popen | None) -> str:
if proc is None:
return ""
output = ""
try:
os.killpg(os.getpgid(proc.pid), signal.SIGTERM)
proc.wait(timeout=5)
except Exception:
try:
os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
except Exception:
pass
if proc.stdout:
try:
output = proc.stdout.read(4000)
except Exception:
output = ""
return output
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--output-json", default=str(DEFAULT_OUT))
parser.add_argument("--screenshot-dir", default=str(HERMES_REPO / ".scratch" / "screenshots" / "canvas"))
parser.add_argument("--plugin-module", default="svrnty_hermes_webui_plugin")
args = parser.parse_args()
if not CANVA_REPO.exists():
raise SystemExit(f"missing canva-editor repo: {CANVA_REPO}")
out_json = Path(args.output_json)
screenshot_dir = Path(args.screenshot_dir)
screenshot_dir.mkdir(parents=True, exist_ok=True)
canva_port = _free_port()
webui_port = _free_port()
canva_base = f"http://127.0.0.1:{canva_port}"
webui_base = f"http://127.0.0.1:{webui_port}"
canva_proc: subprocess.Popen | None = None
webui_proc: subprocess.Popen | None = None
with tempfile.TemporaryDirectory(prefix="hermes-canvas-webui-") as tmp:
tmp_path = Path(tmp)
canva_proc = _start_canva(canva_port, tmp_path / "canva-editor.db")
try:
if not _wait_for(f"{canva_base}/health", timeout=45):
raise RuntimeError(f"canva-editor did not become healthy at {canva_base}/health")
webui_proc = _start_webui(webui_port, tmp_path / "webui-state", canva_base, args.plugin_module)
if not _wait_for(f"{webui_base}/health", timeout=45):
raise RuntimeError(f"webui did not become healthy at {webui_base}/health")
proof = {
"generated_at": datetime.now().isoformat(timespec="seconds"),
"webui_base": webui_base,
"canva_base": canva_base,
"plugin_module": args.plugin_module,
}
cmo_helper = subprocess.run(
[
sys.executable,
str(CMO_REPO / "canvas_command.py"),
"command",
"Create a CMO-triggered draft promo card for Plan B",
"--variants",
"1",
"--brand",
"planb",
"--project",
"Hermes CMO Canvas Smoke",
"--wait",
],
env={**os.environ, "HERMES_WEBUI_URL": webui_base},
capture_output=True,
text=True,
timeout=60,
)
try:
cmo_helper_payload = json.loads(cmo_helper.stdout or "{}")
except ValueError:
cmo_helper_payload = {"parse_error": cmo_helper.stdout}
proof["cmo_helper"] = {
"returncode": cmo_helper.returncode,
"ok": cmo_helper.returncode == 0 and bool(cmo_helper_payload.get("ok")),
"command_id": cmo_helper_payload.get("command_id"),
"screens": len(cmo_helper_payload.get("screens") or []),
"event_wait_ok": bool((cmo_helper_payload.get("event_wait") or {}).get("ok")),
"design_context_status": (cmo_helper_payload.get("design_context") or {}).get("status"),
"secondbrain_status": (cmo_helper_payload.get("design_context") or {}).get("secondbrain_status"),
"stderr": cmo_helper.stderr[-1000:],
}
with sync_playwright() as p:
browser = p.chromium.launch(
headless=True,
executable_path="/usr/bin/google-chrome" if Path("/usr/bin/google-chrome").exists() else None,
args=["--no-sandbox"],
)
page = browser.new_page(viewport={"width": 1440, "height": 1000})
page.goto(webui_base, wait_until="domcontentloaded", timeout=30000)
page.wait_for_function(
"() => window.SvrntyCanvas && window.SvrntyNav && typeof window.switchPanel === 'function'",
timeout=30000,
)
slash_registered = page.evaluate(
"() => typeof COMMANDS !== 'undefined' && COMMANDS.some((cmd) => cmd && cmd.name === 'canvas')"
)
page.evaluate("() => window.switchPanel('canvas')")
page.wait_for_selector("#svrntyCanvasOverlay:not([hidden])", timeout=10000)
page.wait_for_function(
"() => document.querySelector('#svrntyCanvasStatus')?.textContent.includes('online')",
timeout=30000,
)
page.evaluate(
"""() => window.SvrntyCanvas.request({
prompt: 'Create a polished CMO launch dashboard with campaign KPIs',
variants: 2,
brand_id: 'planb',
source: 'visual-smoke'
})"""
)
page.wait_for_function(
"() => document.querySelectorAll('.svrnty-canvas-artboard').length >= 2",
timeout=60000,
)
editable = page.locator('.svrnty-canvas-artboard:first-child .svrnty-canvas-node[contenteditable="true"]').first
editable.fill("Edited KPI panel")
page.click('.svrnty-canvas-artboard:first-child [data-canvas-action="save"]')
try:
page.wait_for_function(
"() => !!window.SvrntyCanvas.state.lastEditPersisted",
timeout=30000,
)
except Exception:
debug_state = page.evaluate(
"""() => ({
artboards: window.SvrntyCanvas.state.artboards.map((b) => ({id:b.id,title:b.title,dirty:b.dirty,badge:b.badge})),
logs: window.SvrntyCanvas.state.log.map((row) => row.message),
editableCount: document.querySelectorAll('.svrnty-canvas-node[contenteditable="true"]').length,
saveButtonCount: document.querySelectorAll('[data-canvas-action="save"]').length
})"""
)
print(json.dumps({"canvas_edit_debug": debug_state}, indent=2), file=sys.stderr)
raise
page.click('.svrnty-canvas-artboard:first-child [data-canvas-action="prototype-source"]')
page.click('.svrnty-canvas-artboard:nth-child(2) [data-canvas-action="prototype-link"]')
page.wait_for_function(
"() => !!window.SvrntyCanvas.state.lastPrototypeEdge",
timeout=30000,
)
page.click('.svrnty-canvas-artboard:first-child [data-canvas-action="export"]')
page.wait_for_function(
"() => window.SvrntyCanvas.state.exportSummary && window.SvrntyCanvas.state.exportSummary.screens.length >= 2",
timeout=30000,
)
state = page.evaluate(
"""(slashRegistered) => ({
connected: window.SvrntyCanvas.state.connected,
artboards: window.SvrntyCanvas.state.artboards.length,
slashCommandRegistered: slashRegistered,
editPersisted: !!window.SvrntyCanvas.state.lastEditPersisted,
prototypeLinked: !!window.SvrntyCanvas.state.lastPrototypeEdge,
exportScreens: window.SvrntyCanvas.state.exportSummary ? window.SvrntyCanvas.state.exportSummary.screens.length : 0,
exportPrototypeEdges: window.SvrntyCanvas.state.exportSummary ? window.SvrntyCanvas.state.exportSummary.prototypeEdges.length : 0,
designContextReady: window.SvrntyCanvas.state.designContext ? window.SvrntyCanvas.state.designContext.status === 'ready' : false,
designContextSource: window.SvrntyCanvas.state.designContext ? window.SvrntyCanvas.state.designContext.source : '',
designContextVersion: window.SvrntyCanvas.state.designContext ? window.SvrntyCanvas.state.designContext.source_version : '',
secondbrainStatus: window.SvrntyCanvas.state.designContext ? window.SvrntyCanvas.state.designContext.secondbrain_status : '',
memoryHintCount: window.SvrntyCanvas.state.designContext && window.SvrntyCanvas.state.designContext.memory_hints ? window.SvrntyCanvas.state.designContext.memory_hints.length : 0,
logs: window.SvrntyCanvas.state.log.map((row) => row.message).slice(0, 8),
status: document.querySelector('#svrntyCanvasStatus')?.textContent || '',
mainClass: document.querySelector('main.main')?.className || ''
})""",
slash_registered,
)
shot = screenshot_dir / "canvas-desktop-proof.png"
page.screenshot(path=str(shot), full_page=True)
browser.close()
proof["state"] = state
proof["pixel_stats"] = _pixel_stats(shot)
proof["pass"] = bool(
state["connected"]
and state["artboards"] >= 2
and state["slashCommandRegistered"]
and state["editPersisted"]
and state["prototypeLinked"]
and state["exportScreens"] >= 2
and state["exportPrototypeEdges"] >= 1
and state["designContextReady"]
and state["designContextSource"] == "cmo.brand_profile_cache"
and state["secondbrainStatus"] in ("ready", "unconfigured")
and state["memoryHintCount"] >= 1
and proof["cmo_helper"]["ok"]
and proof["cmo_helper"]["screens"] >= 1
and proof["cmo_helper"]["event_wait_ok"]
and proof["cmo_helper"]["design_context_status"] == "ready"
and "svrnty-showing-canvas" in state["mainClass"]
and proof["pixel_stats"]["nonblank"]
)
out_json.parent.mkdir(parents=True, exist_ok=True)
out_json.write_text(json.dumps(proof, indent=2), encoding="utf-8")
print(json.dumps(proof, indent=2))
return 0 if proof["pass"] else 1
finally:
webui_log = _stop(webui_proc)
canva_log = _stop(canva_proc)
if webui_log:
print(webui_log, file=sys.stderr)
if canva_log:
print(canva_log, file=sys.stderr)
if __name__ == "__main__":
raise SystemExit(main())

View File

@ -0,0 +1,247 @@
#!/usr/bin/env python3
"""Smoke a natural CMO Hermes turn into the WebUI Canvas bridge.
This proves the runtime handoff JP cares about:
CMO profile -> terminal helper -> Hermes WebUI plugin -> canva-editor -> replayable
Canvas events. The canva-editor is mocked here so the proof isolates routing; the
real Spark qwen3.6 generator is covered by canva-editor's real-model smoke.
"""
from __future__ import annotations
import argparse
import json
import os
import signal
import socket
import subprocess
import sys
import tempfile
import time
from datetime import datetime
from pathlib import Path
from urllib.error import URLError
from urllib.request import urlopen
PLUGIN_REPO = Path(__file__).resolve().parent.parent
HERMES_REPO = PLUGIN_REPO.parent
WEBUI_REPO = HERMES_REPO / "hermes-webui"
AGENT_REPO = HERMES_REPO / "hermes-agent"
CANVA_REPO = HERMES_REPO.parent / "cortex" / "L2-svrnty.tool-canva-editor"
CMO_REPO = HERMES_REPO / "cmo"
DEFAULT_OUT = HERMES_REPO / ".scratch" / "cmo-natural-canvas-proof.json"
def _free_port() -> int:
sock = socket.socket()
sock.bind(("127.0.0.1", 0))
port = sock.getsockname()[1]
sock.close()
return port
def _wait_for(url: str, timeout: int = 45) -> bool:
deadline = time.time() + timeout
while time.time() < deadline:
try:
with urlopen(url, timeout=1) as res:
if res.status < 500:
return True
except URLError:
pass
except Exception:
pass
time.sleep(0.25)
return False
def _fetch_json(url: str, timeout: float = 8.0) -> dict:
with urlopen(url, timeout=timeout) as res:
return json.loads(res.read().decode("utf-8"))
def _start_canva(port: int, db_path: Path) -> subprocess.Popen:
env = os.environ.copy()
env["PATH"] = f"/home/svrnty/sdk/go1.26.0/bin{os.pathsep}{env.get('PATH', '')}"
env["PORT"] = str(port)
env["DB_PATH"] = str(db_path)
return subprocess.Popen(
["go", "run", "./cmd/server", "--mock"],
cwd=CANVA_REPO,
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
preexec_fn=os.setsid,
)
def _start_webui(port: int, state_dir: Path, canva_base: str) -> subprocess.Popen:
env = os.environ.copy()
env.pop("HERMES_WEBUI_PASSWORD", None)
env["CANVA_EDITOR_BASE_URL"] = canva_base
env["HERMES_WEBUI_PYTHON_PLUGIN"] = "svrnty_hermes_webui_plugin"
env["HERMES_WEBUI_PORT"] = str(port)
env["HERMES_WEBUI_STATE_DIR"] = str(state_dir)
env["HERMES_WEBUI_AGENT_DIR"] = str(AGENT_REPO)
env["HERMES_REPO_ROOT"] = str(HERMES_REPO)
env["PYTHONPATH"] = (
f"{PLUGIN_REPO}{os.pathsep}{AGENT_REPO}"
if not env.get("PYTHONPATH")
else f"{PLUGIN_REPO}{os.pathsep}{AGENT_REPO}{os.pathsep}{env['PYTHONPATH']}"
)
py = WEBUI_REPO / "venv" / "bin" / "python"
agent_py = AGENT_REPO / "venv" / "bin" / "python"
python_exe = str(py) if py.exists() else str(agent_py) if agent_py.exists() else "python3"
return subprocess.Popen(
[python_exe, str(WEBUI_REPO / "server.py")],
cwd=AGENT_REPO if AGENT_REPO.exists() else WEBUI_REPO,
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
preexec_fn=os.setsid,
)
def _stop(proc: subprocess.Popen | None) -> str:
if proc is None:
return ""
output = ""
try:
os.killpg(os.getpgid(proc.pid), signal.SIGTERM)
proc.wait(timeout=5)
except Exception:
try:
os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
except Exception:
pass
if proc.stdout:
try:
output = proc.stdout.read(4000)
except Exception:
output = ""
return output
def _run_cmo_turn(webui_base: str, timeout_s: int) -> subprocess.CompletedProcess:
prompt = f"""
You are CMO for Plan B. This is a routing smoke for the Canvas design command center.
Use the terminal tool to run exactly this command:
python3 "{CMO_REPO / "canvas_command.py"}" command "Create one CMO-routed Plan B promo card for a family meal campaign" --variants 1 --brand planb --project "Hermes CMO Natural Smoke" --wait
After the terminal command returns, reply with a compact summary containing command_id, project_id, number of screens, design_context.status, and secondbrain_status. Do not call canva-editor directly.
""".strip()
env = os.environ.copy()
env["HERMES_WEBUI_URL"] = webui_base
env["CMO_LIB"] = str(CMO_REPO)
env["HERMES_INFERENCE_PROVIDER"] = "vllm"
env["HERMES_INFERENCE_MODEL"] = "qwen3.6-35b-a3b"
return subprocess.run(
[
"hermes",
"-p",
"cmo-planb",
"--provider",
"vllm",
"--model",
"qwen3.6-35b-a3b",
"-z",
prompt,
"--toolsets",
"terminal",
"--accept-hooks",
"--yolo",
],
cwd=HERMES_REPO,
env=env,
capture_output=True,
text=True,
timeout=timeout_s,
)
def _canvas_event_summary(webui_base: str) -> dict:
payload = _fetch_json(f"{webui_base}/api/canvas/events?format=json&since=0")
events = payload.get("events") or []
completed = [event for event in events if event.get("type") == "canvas.command.completed"]
persisted = [event for event in events if event.get("type") == "canvas.screen.persisted"]
failed = [event for event in events if event.get("type") == "canvas.command.failed"]
command_ids = sorted({event.get("command_id") for event in completed if event.get("command_id")})
return {
"cursor": payload.get("cursor"),
"event_count": len(events),
"completed_count": len(completed),
"persisted_count": len(persisted),
"failed_count": len(failed),
"command_ids": command_ids,
"last_completed": completed[-1] if completed else None,
"last_persisted": persisted[-1] if persisted else None,
}
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--output-json", default=str(DEFAULT_OUT))
parser.add_argument("--timeout", type=int, default=180)
args = parser.parse_args()
out_json = Path(args.output_json)
canva_port = _free_port()
webui_port = _free_port()
canva_base = f"http://127.0.0.1:{canva_port}"
webui_base = f"http://127.0.0.1:{webui_port}"
canva_proc: subprocess.Popen | None = None
webui_proc: subprocess.Popen | None = None
with tempfile.TemporaryDirectory(prefix="hermes-cmo-canvas-") as tmp:
tmp_path = Path(tmp)
proof = {
"generated_at": datetime.now().isoformat(timespec="seconds"),
"webui_base": webui_base,
"canva_base": canva_base,
"profile": "cmo-planb",
"provider": "vllm",
"model": "qwen3.6-35b-a3b",
}
try:
canva_proc = _start_canva(canva_port, tmp_path / "canva-editor.db")
if not _wait_for(f"{canva_base}/health"):
raise RuntimeError(f"canva-editor did not become healthy at {canva_base}/health")
webui_proc = _start_webui(webui_port, tmp_path / "webui-state", canva_base)
if not _wait_for(f"{webui_base}/health"):
raise RuntimeError(f"webui did not become healthy at {webui_base}/health")
result = _run_cmo_turn(webui_base, args.timeout)
events = _canvas_event_summary(webui_base)
proof["cmo_turn"] = {
"returncode": result.returncode,
"stdout": result.stdout[-4000:],
"stderr": result.stderr[-2000:],
}
proof["events"] = events
proof["pass"] = bool(
result.returncode == 0
and events["completed_count"] >= 1
and events["persisted_count"] >= 1
and events["failed_count"] == 0
and events["command_ids"]
)
out_json.parent.mkdir(parents=True, exist_ok=True)
out_json.write_text(json.dumps(proof, indent=2), encoding="utf-8")
print(json.dumps(proof, indent=2))
return 0 if proof["pass"] else 1
finally:
webui_log = _stop(webui_proc)
canva_log = _stop(canva_proc)
if webui_log:
print(webui_log, file=sys.stderr)
if canva_log:
print(canva_log, file=sys.stderr)
if __name__ == "__main__":
raise SystemExit(main())

254
static/canvas.css Normal file
View File

@ -0,0 +1,254 @@
.svrnty-canvas-overlay {
display: none;
grid-template-rows: auto 1fr;
background: var(--bg, #0b0d10);
color: var(--text, #f5f7fb);
min-width: 0;
overflow: hidden;
}
body > main.svrnty-canvas-overlay {
display: grid;
min-height: 100vh;
}
main.main.svrnty-showing-canvas > .main-view {
display: none !important;
}
main.main.svrnty-showing-canvas > .svrnty-canvas-overlay {
display: grid;
flex: 1 1 100%;
height: 100%;
min-height: 0;
}
.svrnty-canvas-toolbar {
display: flex;
align-items: center;
gap: 10px;
min-height: 56px;
padding: 10px 14px;
border-bottom: 1px solid var(--border2, #252a31);
background: color-mix(in srgb, var(--bg, #0b0d10) 92%, #ffffff 8%);
}
.svrnty-canvas-title {
font-weight: 700;
font-size: 14px;
margin-right: 4px;
}
.svrnty-canvas-status {
font-size: 12px;
color: var(--muted, #9aa4b2);
min-width: 150px;
}
.svrnty-canvas-prompt {
flex: 1;
min-width: 240px;
height: 36px;
border: 1px solid var(--border2, #252a31);
border-radius: 6px;
padding: 0 10px;
background: var(--input-bg, #11151b);
color: inherit;
}
.svrnty-canvas-stepper,
.svrnty-canvas-brand {
height: 36px;
border: 1px solid var(--border2, #252a31);
border-radius: 6px;
padding: 0 8px;
background: var(--input-bg, #11151b);
color: inherit;
}
.svrnty-canvas-btn {
height: 36px;
border: 1px solid var(--border2, #252a31);
border-radius: 6px;
padding: 0 12px;
background: var(--button-bg, #171c23);
color: inherit;
cursor: pointer;
}
.svrnty-canvas-btn.primary {
border-color: #007c57;
background: #007c57;
color: #fff;
}
.svrnty-canvas-btn:disabled {
opacity: .55;
cursor: not-allowed;
}
.svrnty-canvas-body {
display: grid;
grid-template-columns: minmax(0, 1fr) 320px;
min-height: 0;
}
.svrnty-canvas-stage {
position: relative;
overflow: auto;
padding: 28px;
background:
linear-gradient(rgba(255,255,255,.035) 1px, transparent 1px),
linear-gradient(90deg, rgba(255,255,255,.035) 1px, transparent 1px),
#0f1318;
background-size: 32px 32px;
}
.svrnty-canvas-artboards {
display: flex;
align-items: flex-start;
gap: 24px;
min-width: max-content;
}
.svrnty-canvas-artboard {
width: 360px;
min-height: 640px;
border: 1px solid #2b333d;
border-radius: 8px;
background: #f7f8fb;
color: #12161c;
box-shadow: 0 18px 45px rgba(0,0,0,.32);
overflow: hidden;
}
.svrnty-canvas-artboard-head {
display: flex;
justify-content: space-between;
gap: 8px;
padding: 10px 12px;
border-bottom: 1px solid #d8dde5;
background: #fff;
font-size: 12px;
}
.svrnty-canvas-badge {
color: #667085;
white-space: nowrap;
}
.svrnty-canvas-artboard-actions {
display: flex;
gap: 6px;
padding: 8px 10px;
border-bottom: 1px solid #e2e7ef;
background: #fbfcfe;
}
.svrnty-canvas-mini-btn {
min-height: 26px;
border: 1px solid #cbd5e1;
border-radius: 6px;
padding: 0 8px;
background: #fff;
color: #172033;
font-size: 11px;
cursor: pointer;
}
.svrnty-canvas-mini-btn:hover {
border-color: #007c57;
}
.svrnty-canvas-render {
padding: 18px;
}
.svrnty-canvas-node {
border: 1px solid #dde3ec;
border-radius: 6px;
padding: 10px;
margin-bottom: 10px;
background: #fff;
font-size: 13px;
}
.svrnty-canvas-node[contenteditable="true"] {
outline: 1px dashed transparent;
cursor: text;
}
.svrnty-canvas-node[contenteditable="true"]:focus {
outline-color: #007c57;
background: #eefcf6;
}
.svrnty-canvas-node.row {
display: flex;
gap: 8px;
}
.svrnty-canvas-node.button {
display: inline-flex;
background: #007c57;
color: #fff;
border-color: #007c57;
}
.svrnty-canvas-node.text {
border-color: transparent;
background: transparent;
padding: 4px 0;
font-size: 16px;
font-weight: 600;
}
.svrnty-canvas-empty {
color: #a8b3c2;
font-size: 13px;
padding: 18px;
border: 1px dashed #3a4350;
border-radius: 8px;
max-width: 520px;
background: rgba(0,0,0,.16);
}
.svrnty-canvas-rail {
min-width: 0;
border-left: 1px solid var(--border2, #252a31);
background: color-mix(in srgb, var(--bg, #0b0d10) 94%, #ffffff 6%);
overflow: auto;
padding: 14px;
}
.svrnty-canvas-rail h3 {
margin: 0 0 8px;
font-size: 13px;
}
.svrnty-canvas-log {
display: grid;
gap: 8px;
margin-top: 12px;
}
.svrnty-canvas-log-row {
border: 1px solid var(--border2, #252a31);
border-radius: 6px;
padding: 8px;
font-size: 12px;
color: var(--muted, #9aa4b2);
background: rgba(255,255,255,.025);
}
@media (max-width: 900px) {
.svrnty-canvas-toolbar {
flex-wrap: wrap;
}
.svrnty-canvas-body {
grid-template-columns: 1fr;
}
.svrnty-canvas-rail {
display: none;
}
}

39
static/canvas.html Normal file
View File

@ -0,0 +1,39 @@
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Sovereign Canvas</title>
<link rel="stylesheet" href="/plugins/svrnty/canvas.css">
</head>
<body>
<main class="svrnty-canvas-overlay">
<div class="svrnty-canvas-toolbar">
<div class="svrnty-canvas-title">Sovereign Canvas</div>
<div id="standaloneCanvasStatus" class="svrnty-canvas-status">Checking...</div>
</div>
<div class="svrnty-canvas-body">
<div class="svrnty-canvas-stage">
<div class="svrnty-canvas-empty">Open Hermes WebUI and select Canvas for the live command-center panel.</div>
</div>
<aside class="svrnty-canvas-rail">
<h3>Bridge status</h3>
<pre id="standaloneCanvasTools" class="svrnty-canvas-log-row">Loading...</pre>
</aside>
</div>
</main>
<script>
Promise.all([
fetch("/api/canvas/status").then((r) => r.json()),
fetch("/api/canvas/tools").then((r) => r.json())
]).then(([status, tools]) => {
document.getElementById("standaloneCanvasStatus").textContent =
status.ok ? "canva-editor online" : "canva-editor offline";
document.getElementById("standaloneCanvasTools").textContent =
JSON.stringify({ status: status, tools: tools }, null, 2);
}).catch((e) => {
document.getElementById("standaloneCanvasStatus").textContent = String(e);
});
</script>
</body>
</html>

638
static/canvas.js Normal file
View File

@ -0,0 +1,638 @@
// svrnty-canvas: Hermes-native shell for the sovereign Stitch-like design loop.
(function () {
"use strict";
if (window.__svrntyCanvasLoaded) return;
window.__svrntyCanvasLoaded = true;
const state = {
connected: false,
project: null,
artboards: [],
log: [],
eventSource: null,
eventCursor: 0,
eventPoll: null,
prototypeSourceId: null,
exportSummary: null,
lastEditPersisted: null,
lastPrototypeEdge: null,
designContext: null,
};
window.SvrntyCanvas = {
state,
refresh: _refreshStatus,
generate: _generate,
request: _request,
};
function _init() {
window.addEventListener("svrnty:panel-switch", (ev) => {
const name = ev && ev.detail && ev.detail.name;
if (name === "canvas") _openOverlay();
else _closeOverlay();
});
document.addEventListener("svrnty:canvas-generate", (ev) => _request((ev && ev.detail) || {}));
_registerSlashCommand();
const main = document.querySelector("main.main");
if (main && main.classList.contains("svrnty-showing-canvas")) _openOverlay();
}
function _registerSlashCommand() {
const install = () => {
try {
if (typeof COMMANDS === "undefined" || !Array.isArray(COMMANDS)) return false;
if (COMMANDS.some((cmd) => cmd && cmd.name === "canvas")) return true;
COMMANDS.push({
name: "canvas",
desc: "Generate on the sovereign Canvas",
arg: "prompt",
fn: (args) => {
const prompt = String(args || "").trim();
if (!prompt) {
_log("Usage: /canvas <prompt>");
return true;
}
_request({ prompt, source: "slash-command" });
return true;
},
});
return true;
} catch (_) {
return false;
}
};
if (install()) return;
let attempts = 0;
const timer = setInterval(() => {
attempts += 1;
if (install() || attempts > 40) clearInterval(timer);
}, 250);
}
function _openOverlay() {
const mount = document.querySelector("main.main") || document.body;
let overlay = document.getElementById("svrntyCanvasOverlay");
if (!overlay) {
overlay = _buildOverlay();
mount.appendChild(overlay);
} else if (overlay.parentElement !== mount && mount.tagName === "MAIN") {
mount.appendChild(overlay);
}
overlay.hidden = false;
_refreshStatus();
_connectEvents();
_startEventPolling();
_render();
}
function _closeOverlay() {
const overlay = document.getElementById("svrntyCanvasOverlay");
if (overlay) overlay.hidden = true;
if (state.eventSource) {
state.eventSource.close();
state.eventSource = null;
}
if (state.eventPoll) {
clearInterval(state.eventPoll);
state.eventPoll = null;
}
}
function _buildOverlay() {
const root = document.createElement("div");
root.id = "svrntyCanvasOverlay";
root.className = "svrnty-canvas-overlay";
root.innerHTML =
'<div class="svrnty-canvas-toolbar">' +
'<div class="svrnty-canvas-title">Sovereign Canvas</div>' +
'<div id="svrntyCanvasStatus" class="svrnty-canvas-status">Checking...</div>' +
'<select id="svrntyCanvasBrand" class="svrnty-canvas-brand">' +
'<option value="planb">Plan B</option><option value="svrnty">Svrnty</option>' +
'</select>' +
'<input id="svrntyCanvasPrompt" class="svrnty-canvas-prompt" placeholder="Describe the screen CMO should create">' +
'<input id="svrntyCanvasVariants" class="svrnty-canvas-stepper" type="number" min="1" max="6" value="3" title="Variants">' +
'<button id="svrntyCanvasGenerate" class="svrnty-canvas-btn primary" type="button">Generate</button>' +
'<button id="svrntyCanvasRefresh" class="svrnty-canvas-btn" type="button">Refresh</button>' +
'</div>' +
'<div class="svrnty-canvas-body">' +
'<div class="svrnty-canvas-stage"><div id="svrntyCanvasArtboards" class="svrnty-canvas-artboards"></div></div>' +
'<aside class="svrnty-canvas-rail"><h3>Live loop</h3><div id="svrntyCanvasTools"></div><div id="svrntyCanvasLog" class="svrnty-canvas-log"></div></aside>' +
'</div>';
root.querySelector("#svrntyCanvasGenerate").addEventListener("click", _generate);
root.querySelector("#svrntyCanvasRefresh").addEventListener("click", _refreshStatus);
root.querySelector("#svrntyCanvasPrompt").addEventListener("keydown", (ev) => {
if (ev.key === "Enter") _generate();
});
root.querySelector("#svrntyCanvasArtboards").addEventListener("click", _handleArtboardClick);
root.querySelector("#svrntyCanvasArtboards").addEventListener("input", _handleArtboardInput);
return root;
}
function _refreshStatus() {
const brand = document.getElementById("svrntyCanvasBrand");
const brandId = brand ? brand.value : "planb";
Promise.all([
fetch("/api/canvas/status").then((r) => r.json()),
fetch("/api/canvas/tools").then((r) => r.json()),
fetch("/api/canvas/design-context?brand_id=" + encodeURIComponent(brandId)).then((r) => r.json()),
]).then(([status, tools, context]) => {
state.connected = !!(status && status.ok);
state.designContext = context || null;
_setText("svrntyCanvasStatus", state.connected ? "canva-editor online" : "canva-editor offline");
_renderTools(tools.tools || []);
if (status && status.capabilities) _log("Capabilities: " + _capabilitySummary(status.capabilities));
if (context && context.status === "ready") {
_log("BTE context ready: " + (context.source_version || context.fetched_at || context.resolved_brand_id));
} else if (context && context.deferred_ports) {
_log("Context degraded: " + context.deferred_ports.join(", "));
}
_log(state.connected ? "Canvas bridge online" : "Canvas bridge waiting for canva-editor");
if (state.connected) _ensureProject().catch((e) => _log("Project load failed: " + _message(e)));
}).catch((e) => {
state.connected = false;
_setText("svrntyCanvasStatus", "canvas bridge error");
_log("Status failed: " + _message(e));
});
}
function _connectEvents() {
if (!window.EventSource || state.eventSource) return;
try {
state.eventSource = new EventSource("/api/canvas/events");
state.eventSource.addEventListener("canvas.connected", () => _log("Live event stream connected"));
state.eventSource.onerror = () => {
if (state.eventSource) {
state.eventSource.close();
state.eventSource = null;
}
};
} catch (_) {}
}
function _startEventPolling() {
if (state.eventPoll) return;
_pollEvents();
state.eventPoll = setInterval(_pollEvents, 1200);
}
function _pollEvents() {
fetch("/api/canvas/events?format=json&since=" + encodeURIComponent(state.eventCursor))
.then((r) => r.json())
.then((body) => {
const events = body && Array.isArray(body.events) ? body.events : [];
events.forEach(_applyEvent);
if (body && body.cursor != null) state.eventCursor = Math.max(state.eventCursor, body.cursor);
})
.catch(() => {});
}
function _applyEvent(ev) {
if (!ev || !ev.type) return;
if (ev.id != null) state.eventCursor = Math.max(state.eventCursor, ev.id);
if (ev.type === "canvas.command.accepted") _log("CMO command accepted: " + (ev.prompt || "Canvas command"));
else if (ev.type === "canvas.project.ready") _log("Project ready: " + ((ev.project && ev.project.name) || "Hermes Canvas"));
else if (ev.type === "canvas.variant.started") _log("Generating variant " + ev.variant_index + "/" + ev.variants);
else if (ev.type === "canvas.screen.persisted") {
_ingestScreenEvent(ev);
_log("Artboard persisted from command");
} else if (ev.type === "canvas.command.completed") _log("Canvas command completed");
else if (ev.type === "canvas.command.failed") _log("Canvas command failed: " + (ev.error || "unknown error"));
}
function _generate() {
const input = document.getElementById("svrntyCanvasPrompt");
const prompt = input ? input.value.trim() : "";
if (!prompt) {
_log("Prompt required before generation");
return;
}
const btn = document.getElementById("svrntyCanvasGenerate");
if (btn) btn.disabled = true;
_log("CMO request queued: " + prompt);
const variantsInput = document.getElementById("svrntyCanvasVariants");
const variants = Math.max(1, Math.min(6, parseInt((variantsInput && variantsInput.value) || "1", 10)));
const brand = document.getElementById("svrntyCanvasBrand");
const brandId = brand ? brand.value : "planb";
_runCommand(prompt, variants, brandId).finally(() => {
if (btn) btn.disabled = false;
});
}
function _request(payload) {
const prompt = String((payload && payload.prompt) || "").trim();
if (!prompt) {
_log("Canvas request ignored: prompt required");
return false;
}
try {
if (typeof window.switchPanel === "function") {
window.switchPanel("canvas", { fromRailClick: true });
} else {
_openOverlay();
}
} catch (_) {
_openOverlay();
}
setTimeout(() => {
const input = document.getElementById("svrntyCanvasPrompt");
const variantsInput = document.getElementById("svrntyCanvasVariants");
const brand = document.getElementById("svrntyCanvasBrand");
if (input) input.value = prompt;
if (variantsInput && payload.variants) variantsInput.value = String(payload.variants);
if (brand && (payload.brand_id || payload.brandId)) brand.value = payload.brand_id || payload.brandId;
_log("Canvas request from " + (payload.source || "Hermes") + ": " + prompt);
_generate();
}, 0);
return true;
}
async function _runCommand(prompt, variants, brandId) {
try {
const r = await fetch("/api/canvas/command", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ prompt, variants, brand_id: brandId }),
});
const body = await r.json();
if (!r.ok) throw new Error(body && body.error ? body.error : "HTTP " + r.status);
if (body.project) state.project = body.project;
if (body.design_context) state.designContext = body.design_context;
(body.events || []).forEach(_applyEvent);
(body.screens || []).forEach((item) => _ingestScreenEvent({
screen: item.screen,
screenSpec: item.screenSpec,
prompt: item.prompt || prompt,
variant_index: item.variant_index,
}));
_render();
} catch (e) {
_log("Command failed: " + _message(e));
}
}
async function _generateVariants(prompt, variants) {
for (let i = 0; i < variants; i++) {
const variantPrompt = variants === 1 ? prompt : prompt + "\nVariant " + (i + 1) + " of " + variants + ": choose a distinct composition.";
try {
_log("Generating variant " + (i + 1) + "/" + variants);
const r = await fetch("/api/canvas/proxy?path=/api/v1/generate", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ prompt: variantPrompt }),
});
const body = await r.json();
if (!r.ok) throw new Error(body && body.error ? JSON.stringify(body.error) : "HTTP " + r.status);
const spec = body.screenSpec || body.screen_spec || body;
const saved = await _saveScreen(prompt, spec).catch((e) => {
_log("Generated but not persisted: " + _message(e));
return null;
});
state.artboards.unshift(_artboardFromSpec(
saved,
spec,
prompt.slice(0, 42) || "Generated screen",
variants === 1 ? "generated" : "variant " + (i + 1),
));
_log("Artboard generated");
_render();
} catch (e) {
_log("Generation failed: " + _message(e));
}
}
}
async function _ensureProject() {
if (state.project) return state.project;
const projects = await _canvasFetch("/api/v1/projects", { method: "GET" });
state.project = Array.isArray(projects) && projects.length
? projects[0]
: await _canvasFetch("/api/v1/projects", {
method: "POST",
body: { name: "Hermes Canvas" },
});
_log("Project loaded: " + (state.project.name || state.project.id || "Hermes Canvas"));
await _loadScreens();
return state.project;
}
async function _loadScreens() {
if (!state.project) return;
const projectId = state.project.id || state.project.ID || state.project.uuid;
if (!projectId) return;
const screens = await _canvasFetch("/api/v1/projects/" + encodeURIComponent(projectId) + "/screens", { method: "GET" });
state.artboards = (Array.isArray(screens) ? screens : []).map((screen, idx) =>
_artboardFromSpec(
screen,
screen.spec || screen.Spec || screen.screenSpec || {},
screen.name || screen.Name || "Screen " + (idx + 1),
"saved",
)
);
_render();
}
async function _saveScreen(prompt, spec) {
await _ensureProject();
const projectId = state.project.id || state.project.ID || state.project.uuid;
if (!projectId) throw new Error("project id missing");
return _canvasFetch("/api/v1/projects/" + encodeURIComponent(projectId) + "/screens", {
method: "POST",
body: {
name: prompt.slice(0, 42) || "Generated screen",
spec,
},
});
}
async function _canvasFetch(path, options) {
const upstreamMethod = (options && options.method) || "GET";
const bridgeMethod = upstreamMethod === "PUT" ? "POST" : upstreamMethod;
const opts = {
method: bridgeMethod,
headers: {},
};
if (options && options.body !== undefined) {
opts.headers["Content-Type"] = "application/json";
opts.body = JSON.stringify(options.body);
}
const methodQuery = upstreamMethod === "PUT" ? "&method=PUT" : "";
const r = await fetch("/api/canvas/proxy?path=" + encodeURIComponent(path) + methodQuery, opts);
const text = await r.text();
const body = text ? JSON.parse(text) : {};
if (!r.ok) throw new Error(body && body.error ? JSON.stringify(body.error) : "HTTP " + r.status);
return body;
}
function _artboardFromSpec(record, spec, title, badge) {
return {
id: record && (record.id || record.ID || record.uuid),
title,
badge: record && (record.id || record.ID || record.uuid) ? badge + " · persisted" : badge,
spec,
dirty: false,
};
}
function _ingestScreenEvent(ev) {
const screen = ev.screen || {};
const spec = ev.screenSpec || screen.spec || screen.Spec || {};
const id = screen.id || screen.ID || screen.uuid;
if (id && state.artboards.some((board) => board.id === id)) return;
state.artboards.unshift(_artboardFromSpec(
screen,
spec,
screen.name || screen.Name || (ev.prompt || "Generated screen").slice(0, 42),
ev.variant_index ? "command variant " + ev.variant_index : "command",
));
_render();
}
function _render() {
const target = document.getElementById("svrntyCanvasArtboards");
if (!target) return;
if (!state.artboards.length) {
target.innerHTML = '<div class="svrnty-canvas-empty">Prompt CMO to generate a screen. Artboards, variants, edits, prototype links, and exports will accumulate here.</div>';
return;
}
target.innerHTML = state.artboards.map((board, idx) =>
'<article class="svrnty-canvas-artboard">' +
'<div class="svrnty-canvas-artboard-head"><strong>' + _esc(board.title) + '</strong><span class="svrnty-canvas-badge">' + _esc(board.dirty ? "edited" : board.badge) + '</span></div>' +
'<div class="svrnty-canvas-artboard-actions">' +
'<button class="svrnty-canvas-mini-btn" type="button" data-canvas-action="save" data-board-index="' + idx + '">Save edits</button>' +
'<button class="svrnty-canvas-mini-btn" type="button" data-canvas-action="prototype-source" data-board-index="' + idx + '">Start</button>' +
'<button class="svrnty-canvas-mini-btn" type="button" data-canvas-action="prototype-link" data-board-index="' + idx + '">Link</button>' +
'<button class="svrnty-canvas-mini-btn" type="button" data-canvas-action="export" data-board-index="' + idx + '">Export</button>' +
'</div>' +
'<div class="svrnty-canvas-render">' + _renderSpec(board.spec) + '</div>' +
'</article>'
).join("");
}
function _renderSpec(spec) {
if (!spec) return '<div class="svrnty-canvas-node">Empty spec</div>';
const root = _rootRef(spec);
return _renderNode(root.node, 0, root.path);
}
function _renderNode(node, depth, path) {
if (!node || typeof node !== "object" || depth > 12) return "";
const type = String(node.type || node.component || "container").toLowerCase();
const field = _editableField(node);
const text = field ? _editableValue(node, field) : (node.text || node.label || node.title || node.name || (node.props && (node.props.data || node.props.label || node.props.title)) || type);
const children = node.children || node.items || [];
const childKey = Array.isArray(node.children) ? "children" : Array.isArray(node.items) ? "items" : "";
const childHtml = Array.isArray(children)
? children.map((child, idx) => _renderNode(child, depth + 1, path.concat([childKey, String(idx)]))).join("")
: "";
const editAttrs = field
? ' contenteditable="true" spellcheck="false" data-canvas-path="' + _esc(path.join(".")) + '" data-canvas-field="' + _esc(field) + '"'
: "";
return '<div class="svrnty-canvas-node ' + _esc(type) + '"' + editAttrs + '>' + _esc(text) + childHtml + '</div>';
}
function _handleArtboardInput(ev) {
const el = ev.target && ev.target.closest && ev.target.closest("[data-canvas-path]");
if (!el) return;
const boardEl = el.closest(".svrnty-canvas-artboard");
const boardIndex = Array.prototype.indexOf.call(boardEl.parentElement.children, boardEl);
const board = state.artboards[boardIndex];
if (!board) return;
const field = el.getAttribute("data-canvas-field");
const node = _nodeAtPath(board.spec, (el.getAttribute("data-canvas-path") || "").split(".").filter(Boolean));
if (!node || !field) return;
_setEditableValue(node, field, el.childNodes.length ? Array.prototype.filter.call(el.childNodes, (n) => n.nodeType === Node.TEXT_NODE).map((n) => n.nodeValue).join("").trim() || el.textContent.trim() : el.textContent.trim());
board.dirty = true;
const badge = boardEl.querySelector(".svrnty-canvas-badge");
if (badge) badge.textContent = "edited";
}
function _handleArtboardClick(ev) {
const btn = ev.target && ev.target.closest && ev.target.closest("[data-canvas-action]");
if (!btn) return;
const board = state.artboards[parseInt(btn.getAttribute("data-board-index") || "-1", 10)];
if (!board) return;
const action = btn.getAttribute("data-canvas-action");
if (action === "save") _persistBoard(board);
else if (action === "prototype-source") _setPrototypeSource(board);
else if (action === "prototype-link") _linkPrototype(board);
else if (action === "export") _exportProject();
}
async function _persistBoard(board) {
_log("Saving simple edit: " + (board.title || board.id || "artboard"));
if (!board.id) {
_log("Save skipped: artboard has no screen id");
return;
}
await _ensureProject();
const projectId = state.project && (state.project.id || state.project.ID || state.project.uuid);
if (!projectId) {
_log("Save failed: project id missing");
return;
}
try {
const saved = await _canvasFetch(
"/api/v1/projects/" + encodeURIComponent(projectId) + "/screens/" + encodeURIComponent(board.id),
{ method: "PUT", body: { spec: board.spec } },
);
board.spec = saved.spec || saved.Spec || board.spec;
board.dirty = false;
board.badge = "edited · persisted";
state.lastEditPersisted = { screenId: board.id, at: Date.now() };
_log("Simple edit persisted: " + (board.title || board.id));
_render();
} catch (e) {
_log("Save failed: " + _message(e));
}
}
function _setPrototypeSource(board) {
if (!board.id) {
_log("Prototype start needs a persisted screen");
return;
}
state.prototypeSourceId = board.id;
_log("Prototype start set: " + (board.title || board.id));
}
async function _linkPrototype(board) {
if (!state.prototypeSourceId) {
_setPrototypeSource(board);
return;
}
if (!board.id || board.id === state.prototypeSourceId) {
_log("Choose a different persisted screen to link");
return;
}
await _ensureProject();
const projectId = state.project && (state.project.id || state.project.ID || state.project.uuid);
if (!projectId) {
_log("Prototype link failed: project id missing");
return;
}
try {
const edge = await _canvasFetch(
"/api/v1/projects/" + encodeURIComponent(projectId) + "/prototype-edges",
{
method: "POST",
body: {
sourceScreenId: state.prototypeSourceId,
targetScreenId: board.id,
action: "navigate",
},
},
);
state.lastPrototypeEdge = edge;
_log("Prototype link saved: " + (edge.sourceScreenId || state.prototypeSourceId) + " -> " + (edge.targetScreenId || board.id));
state.prototypeSourceId = board.id;
} catch (e) {
_log("Prototype link failed: " + _message(e));
}
}
async function _exportProject() {
await _ensureProject();
const projectId = state.project && (state.project.id || state.project.ID || state.project.uuid);
if (!projectId) {
_log("Export failed: project id missing");
return null;
}
try {
const exported = await _canvasFetch("/api/v1/projects/" + encodeURIComponent(projectId) + "/export", { method: "GET" });
state.exportSummary = exported;
const screens = (exported.screens || []).length;
const variants = (exported.variants || []).length;
const edges = (exported.prototypeEdges || []).length;
_log("Export ready: " + screens + " screens, " + variants + " variants, " + edges + " prototype links");
return exported;
} catch (e) {
_log("Export failed: " + _message(e));
return null;
}
}
function _rootRef(spec) {
if (spec && spec.screen && spec.screen.root) return { node: spec.screen.root, path: ["screen", "root"] };
if (spec && spec.root) return { node: spec.root, path: ["root"] };
if (spec && spec.widget) return { node: spec.widget, path: ["widget"] };
return { node: spec, path: [] };
}
function _nodeAtPath(root, path) {
return path.reduce((node, part) => node && node[part], root);
}
function _editableField(node) {
if (!node || typeof node !== "object") return "";
if (Array.isArray(node.children) || Array.isArray(node.items)) return "";
const direct = ["text", "label", "title", "name"].find((field) => typeof node[field] === "string");
if (direct) return direct;
if (node.props && typeof node.props === "object") {
const prop = ["data", "label", "title", "subtitle", "description"].find((field) => typeof node.props[field] === "string");
if (prop) return "props." + prop;
}
return "";
}
function _editableValue(node, field) {
if (field.indexOf(".") === -1) return node[field];
return field.split(".").reduce((cur, key) => cur && cur[key], node);
}
function _setEditableValue(node, field, value) {
const parts = field.split(".");
const key = parts.pop();
const target = parts.reduce((cur, part) => cur && cur[part], node);
if (target && key) target[key] = value;
}
function _renderTools(tools) {
const target = document.getElementById("svrntyCanvasTools");
if (!target) return;
target.innerHTML = (tools || []).slice(0, 6).map((tool) =>
'<div class="svrnty-canvas-log-row"><strong>' + _esc(tool.name) + '</strong><br>' + _esc(tool.purpose || "") + '</div>'
).join("");
}
function _log(message) {
state.log.unshift({ message, at: new Date().toLocaleTimeString() });
state.log = state.log.slice(0, 8);
const target = document.getElementById("svrntyCanvasLog");
if (!target) return;
target.innerHTML = state.log.map((row) =>
'<div class="svrnty-canvas-log-row"><strong>' + _esc(row.at) + '</strong><br>' + _esc(row.message) + '</div>'
).join("");
}
function _setText(id, text) {
const el = document.getElementById(id);
if (el) el.textContent = text;
}
function _message(e) {
return e && e.message ? e.message : String(e);
}
function _esc(s) {
return String(s == null ? "" : s).replace(/[<>&"]/g, (c) => ({
"<": "&lt;",
">": "&gt;",
"&": "&amp;",
'"': "&quot;",
}[c]));
}
function _capabilitySummary(caps) {
const advanced = ["variants", "promptEdit", "prototypeGraph", "liveEvents"]
.filter((key) => caps[key])
.join(", ");
return advanced || "base Stitch loop";
}
if (document.readyState === "loading") {
document.addEventListener("DOMContentLoaded", _init);
} else {
_init();
}
})();

View File

@ -0,0 +1,109 @@
.cortex-os-runtime-health {
border: 1px solid rgba(19, 82, 121, 0.22);
border-radius: 8px;
margin: 16px;
max-width: 520px;
padding: 14px 16px;
background: #f7fbff;
color: #18212b;
font: 14px/1.45 system-ui, -apple-system, BlinkMacSystemFont, "Segoe UI", sans-serif;
}
main.main:not(.svrnty-showing-cortex-os) .cortex-os-runtime-health {
display: none;
}
.cortex-os-runtime-health h2 {
margin: 0 0 8px;
font-size: 15px;
font-weight: 700;
}
.cortex-os-runtime-health__badge {
display: inline-flex;
align-items: center;
min-height: 22px;
padding: 0 8px;
border-radius: 999px;
background: #dde8f0;
color: #18212b;
font-size: 12px;
font-weight: 700;
text-transform: uppercase;
}
.cortex-os-runtime-health[data-state="healthy"] .cortex-os-runtime-health__badge {
background: #d8f1df;
color: #0d5b27;
}
.cortex-os-runtime-health[data-state="degraded"] .cortex-os-runtime-health__badge,
.cortex-os-runtime-health[data-state="unknown"] .cortex-os-runtime-health__badge,
.cortex-os-runtime-health[data-state="loading"] .cortex-os-runtime-health__badge {
background: #fff0c9;
color: #6a4a00;
}
.cortex-os-runtime-health[data-state="unavailable"] .cortex-os-runtime-health__badge,
.cortex-os-runtime-health[data-state="error"] .cortex-os-runtime-health__badge,
.cortex-os-runtime-health[data-state="redacted"] .cortex-os-runtime-health__badge {
background: #ffe0dd;
color: #7a1e17;
}
.cortex-os-runtime-health__summary {
margin: 10px 0;
}
.cortex-os-runtime-health__signals {
display: grid;
grid-template-columns: minmax(120px, 1fr) minmax(92px, auto);
gap: 6px 12px;
margin: 0;
}
.cortex-os-runtime-health__signals dt,
.cortex-os-runtime-health__signals dd {
margin: 0;
}
.cortex-os-runtime-health__signals dt {
color: #44515f;
}
.cortex-os-runtime-health__signals dd {
font-weight: 700;
}
.cortex-os-runtime-health__details {
border-top: 1px solid rgba(19, 82, 121, 0.14);
display: grid;
gap: 10px;
margin-top: 12px;
padding-top: 12px;
}
.cortex-os-runtime-health__detail-group h3 {
margin: 0 0 4px;
font-size: 12px;
font-weight: 700;
text-transform: uppercase;
}
.cortex-os-runtime-health__detail-list {
display: grid;
grid-template-columns: minmax(116px, 0.8fr) minmax(0, 1.2fr);
gap: 4px 10px;
margin: 0;
}
.cortex-os-runtime-health__detail-list dt,
.cortex-os-runtime-health__detail-list dd {
margin: 0;
min-width: 0;
overflow-wrap: anywhere;
}
.cortex-os-runtime-health__detail-list dt {
color: #44515f;
}

View File

@ -0,0 +1,206 @@
(function () {
"use strict";
var endpoint = "/api/cortex-os/runtime-health";
var states = ["healthy", "degraded", "unavailable", "unknown", "loading", "error", "redacted"];
if (window.__cortexOsRuntimeHealthLoaded) {
return;
}
window.__cortexOsRuntimeHealthLoaded = true;
function ensurePanel() {
var panel = document.querySelector("[data-cortex-os-runtime-health]");
if (panel) {
return panel;
}
panel = document.createElement("section");
panel.className = "cortex-os-runtime-health";
panel.setAttribute("data-cortex-os-runtime-health", "true");
panel.setAttribute("data-state", "loading");
var title = document.createElement("h2");
title.setAttribute("data-role", "headline");
title.textContent = "Cortex OS Runtime Health";
var badge = document.createElement("span");
badge.className = "cortex-os-runtime-health__badge";
badge.setAttribute("data-role", "status");
badge.textContent = "loading";
var summary = document.createElement("p");
summary.className = "cortex-os-runtime-health__summary";
summary.setAttribute("data-role", "summary");
summary.textContent = "Checking Runtime Health.";
var list = document.createElement("dl");
list.className = "cortex-os-runtime-health__signals";
list.setAttribute("data-role", "signals");
var details = document.createElement("section");
details.className = "cortex-os-runtime-health__details";
details.setAttribute("data-role", "details");
panel.appendChild(title);
panel.appendChild(badge);
panel.appendChild(summary);
panel.appendChild(list);
panel.appendChild(details);
var target = document.querySelector("main.main");
if (!target) {
return panel;
}
target.appendChild(panel);
return panel;
}
function setState(panel, state, headline, detail, signals, payloadDetails) {
var nextState = states.indexOf(state) >= 0 ? state : "unknown";
panel.setAttribute("data-state", nextState);
panel.querySelector("[data-role='headline']").textContent = headline || "Runtime Health";
panel.querySelector("[data-role='status']").textContent = nextState;
panel.querySelector("[data-role='summary']").textContent = detail || "redacted";
renderSignals(panel.querySelector("[data-role='signals']"), signals || []);
renderDetails(panel.querySelector("[data-role='details']"), payloadDetails || {});
}
function renderSignals(list, signals) {
list.textContent = "";
signals.slice(0, 3).forEach(function (signal) {
var name = document.createElement("dt");
var value = document.createElement("dd");
name.textContent = signal.signal_id || "redacted";
value.textContent = signal.state || "unknown";
if (signal.summary) {
value.setAttribute("title", signal.summary);
}
list.appendChild(name);
list.appendChild(value);
});
}
function renderKeyValues(container, title, rows) {
var group = document.createElement("section");
group.className = "cortex-os-runtime-health__detail-group";
var heading = document.createElement("h3");
heading.textContent = title;
group.appendChild(heading);
var list = document.createElement("dl");
list.className = "cortex-os-runtime-health__detail-list";
rows.forEach(function (row) {
var key = document.createElement("dt");
var value = document.createElement("dd");
key.textContent = row.key || "redacted";
value.textContent = row.value || "redacted";
list.appendChild(key);
list.appendChild(value);
});
group.appendChild(list);
container.appendChild(group);
}
function renderDetails(container, detail) {
container.textContent = "";
var signals = detail.signals || [];
var warnings = detail.warnings || [];
var redactions = detail.redactions || [];
var sourceTrace = detail.source_trace || [];
var authority = detail.authority || {};
var allAuthorityDenied = Object.keys(authority).every(function (key) {
return authority[key] === false;
});
renderKeyValues(container, "Signals", signals.slice(0, 3).map(function (signal) {
return {
key: signal.signal_id || "redacted",
value: (signal.state || "unknown") + " | " + (signal.redacted ? "redacted" : "visible") + " | " + (signal.summary || "redacted")
};
}));
renderKeyValues(container, "Warnings", warnings.slice(0, 3).map(function (warning) {
return {
key: warning.code || "redacted",
value: warning.message || "redacted"
};
}));
renderKeyValues(container, "Redactions", redactions.slice(0, 4).map(function (redaction) {
return {
key: redaction.reason_code || "redacted",
value: redaction.label || "redacted"
};
}));
renderKeyValues(container, "Source Trace", sourceTrace.slice(0, 3).map(function (source) {
return {
key: source.source_kind || "redacted",
value: source.label || "redacted"
};
}));
renderKeyValues(container, "Authority", [{
key: "all denied",
value: allAuthorityDenied ? "true" : "review required"
}]);
}
function renderPayload(panel, payload) {
if (!payload || payload.ok !== true || !payload.result) {
setState(panel, "error", "Runtime Health is unavailable.", "Runtime Health is unavailable.", []);
return;
}
var displaySummary = payload.result.display_summary || {};
setState(
panel,
payload.result.status,
displaySummary.headline,
displaySummary.detail,
payload.result.signals,
{
signals: payload.result.signals,
warnings: payload.result.warnings,
redactions: payload.result.redactions,
source_trace: payload.result.source_trace,
authority: payload.result.authority
}
);
}
function loadRuntimeHealth() {
var panel = ensurePanel();
setState(panel, "loading", "Cortex OS Runtime Health", "Checking Runtime Health.", []);
fetch(endpoint, {
method: "GET",
headers: { Accept: "application/json" },
credentials: "same-origin",
cache: "no-store"
})
.then(function (response) {
if (!response.ok) {
throw new Error("runtime-health-unavailable");
}
return response.json();
})
.then(function (payload) {
renderPayload(panel, payload);
})
.catch(function () {
setState(panel, "error", "Runtime Health is unavailable.", "Runtime Health is unavailable.", []);
});
}
function handlePanelSwitch(event) {
var detail = (event && event.detail) || {};
if (detail.name === "cortex-os") {
loadRuntimeHealth();
}
}
window.addEventListener("svrnty:panel-switch", handlePanelSwitch);
var main = document.querySelector("main.main");
if (main && main.classList.contains("svrnty-showing-cortex-os")) {
loadRuntimeHealth();
}
})();

View File

@ -22,10 +22,16 @@
'<circle cx="12" cy="12" r="10"/><circle cx="12" cy="12" r="6"/><circle cx="12" cy="12" r="2"/>',
bte:
'<path d="M12 2l1.8 5.6L19.4 9.4l-4.5 3.3 1.7 5.7L12 15l-4.6 3.4 1.7-5.7L4.6 9.4l5.6-1.8L12 2z"/>',
canvas:
'<rect x="3" y="3" width="18" height="18" rx="2"/><path d="M8 3v18M16 3v18M3 8h18M3 16h18"/>',
"cortex-os":
'<path d="M4 7h16M4 12h16M4 17h10"/><circle cx="17" cy="17" r="3"/>',
};
const TABS = [
{ id: "adwright", label: "Adwright", tooltip: "Adwright — marketing intelligence" },
{ id: "bte", label: "BTE", tooltip: "BTE — brand creative studio" },
{ id: "canvas", label: "Canvas", tooltip: "Canvas — sovereign UI design" },
{ id: "cortex-os", label: "Cortex OS", tooltip: "Cortex OS — Runtime Health" },
];
function _svg(iconPath, size, stroke) {

View File

@ -5,8 +5,10 @@ contract still holds after upstream changes. Minimal by design (per protocol
decision Q3): catch gross breakage, evolve as issues surface.
"""
from pathlib import Path
import sys
ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT))
def test_eval_loader_contract_unchanged():
@ -78,6 +80,43 @@ def test_eval_brand_skin_url_contract():
plugin.register(api)
api.inject_stylesheet.assert_any_call("/plugins/svrnty/app.css")
api.inject_script.assert_any_call("/plugins/svrnty/app.js")
api.inject_stylesheet.assert_any_call("/plugins/svrnty/canvas.css")
api.inject_script.assert_any_call("/plugins/svrnty/canvas.js")
def test_eval_canvas_tool_surface_contract():
"""Canvas bridge exposes the first CMO-facing Stitch-like tool contract."""
import json
from routes import canvas
class _H:
def __init__(self):
self.body = b""
self.headers = {}
def send_response(self, c): self.status = c
def send_header(self, k, v): self.headers[k] = v
def end_headers(self): pass
@property
def wfile(self):
h = self
class _W:
def write(self_, b): h.body += b
return _W()
h = _H()
canvas._handle_tools(h, None)
payload = json.loads(h.body)
names = {tool["name"] for tool in payload["tools"]}
assert "canvas.create_project" in names
assert "canvas.command" in names
assert "canvas.generate_screen" in names
assert "canvas.design_context" in names
assert "canvas.update_screen" in names
assert "canvas.record_variant" in names
assert "canvas.connect_prototype_edge" in names
assert "canvas.export_project" in names
def test_eval_connection_map_has_no_forced_internals():

View File

@ -13,6 +13,7 @@ import pytest
PLUGIN_REPO = Path(__file__).resolve().parents[2]
FORK_REPO = PLUGIN_REPO.parent / "hermes-webui"
MANIFEST = PLUGIN_REPO / "manifest.yaml"
@pytest.fixture(scope="module")
@ -60,18 +61,40 @@ def test_loader_register_wires_our_plugin(loader, monkeypatch):
# Core routes registered, including the /umbrella graph API pair.
assert ("POST", "/api/transcribe") in loader._ROUTES
assert ("GET", "/api/vault/status") in loader._ROUTES
assert ("GET", "/api/canvas/status") in loader._ROUTES
assert ("GET", "/api/canvas/tools") in loader._ROUTES
assert ("GET", "/api/canvas/proxy") in loader._ROUTES
assert ("POST", "/api/canvas/proxy") in loader._ROUTES
assert ("PUT", "/api/canvas/proxy") in loader._ROUTES
assert ("POST", "/api/canvas/command") in loader._ROUTES
assert ("GET", "/api/canvas/design-context") in loader._ROUTES
assert ("GET", "/api/canvas/events") in loader._ROUTES
assert ("GET", "/api/umbrella") in loader._ROUTES
assert ("GET", "/api/umbrella/doc") in loader._ROUTES
assert ("GET", "/api/cortex-os/runtime-health") in loader._ROUTES
# Static + injected URLs
assert "svrnty" in loader._STATIC
assert "/plugins/svrnty/app.css" in loader._STYLESHEETS
assert "/plugins/svrnty/app.js" in loader._SCRIPTS
assert "/plugins/svrnty/canvas.css" in loader._STYLESHEETS
assert "/plugins/svrnty/canvas.js" in loader._SCRIPTS
assert "/plugins/svrnty/umbrella_inline.css" in loader._STYLESHEETS
assert "/plugins/svrnty/umbrella_inline.js" in loader._SCRIPTS
assert "/plugins/svrnty/cortex-os/runtime-health/runtime_health.css" in loader._STYLESHEETS
assert "/plugins/svrnty/cortex-os/runtime-health/runtime_health.js" in loader._SCRIPTS
# Audio processor for voice-message transcription
assert len(loader._AUDIO_PROCESSORS) == 1
def test_manifest_declares_cortex_os_runtime_health_surface():
manifest = MANIFEST.read_text(encoding="utf-8")
assert "/api/cortex-os/runtime-health" in manifest
assert "routes/cortex_os_extension.py" in manifest
assert "/plugins/svrnty/cortex-os/runtime-health/runtime_health.css" in manifest
assert "/plugins/svrnty/cortex-os/runtime-health/runtime_health.js" in manifest
def test_loader_noop_when_env_unset(loader, monkeypatch):
"""No env var = no plugin loaded. Upstream behavior fully preserved."""
monkeypatch.delenv("HERMES_WEBUI_PYTHON_PLUGIN", raising=False)

327
tests/unit/test_canvas.py Normal file
View File

@ -0,0 +1,327 @@
"""Unit tests for the Canvas bridge route."""
import json
import sqlite3
import sys
from types import SimpleNamespace
from pathlib import Path
from unittest.mock import MagicMock, patch
ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT))
from routes import canvas
class _Handler:
command = "GET"
def __init__(self, body=b""):
self.status = None
self.headers_out = {}
self.body = b""
self.headers = {
"Content-Type": "application/json",
"Content-Length": str(len(body)),
}
self.rfile = _Reader(body)
def send_response(self, status):
self.status = status
def send_header(self, key, value):
self.headers_out[key] = value
def end_headers(self):
pass
@property
def wfile(self):
outer = self
class _W:
def write(self, body):
outer.body += body
return _W()
class _Reader:
def __init__(self, body):
self._body = body
def read(self, _length):
return self._body
def _parsed(query):
return SimpleNamespace(query=query)
def test_register_wires_canvas_routes():
api = MagicMock()
canvas.register(api)
calls = {(c.args[1], c.args[0]) for c in api.register_route.call_args_list}
assert ("GET", "/api/canvas/status") in calls
assert ("GET", "/api/canvas/proxy") in calls
assert ("POST", "/api/canvas/proxy") in calls
assert ("PUT", "/api/canvas/proxy") in calls
assert ("POST", "/api/canvas/command") in calls
assert ("GET", "/api/canvas/tools") in calls
assert ("GET", "/api/canvas/design-context") in calls
assert ("GET", "/api/canvas/events") in calls
def test_proxy_blocks_unlisted_path():
h = _Handler()
assert canvas._handle_proxy(h, _parsed("path=/api/v1/admin")) is True
assert h.status == 403
assert json.loads(h.body)["error"] == "path not allowed: /api/v1/admin"
def test_proxy_forwards_allowed_projects_path():
h = _Handler()
response = MagicMock()
response.__enter__.return_value.status = 200
response.__enter__.return_value.headers = {"Content-Type": "application/json"}
response.__enter__.return_value.read.return_value = b"[]"
with patch("routes.canvas.urllib.request.urlopen", return_value=response) as urlopen:
assert canvas._handle_proxy(h, _parsed("path=/api/v1/projects")) is True
assert h.status == 200
assert h.body == b"[]"
assert urlopen.call_args.args[0].full_url.endswith("/api/v1/projects")
def test_proxy_forwards_capabilities_path():
h = _Handler()
response = MagicMock()
response.__enter__.return_value.status = 200
response.__enter__.return_value.headers = {"Content-Type": "application/json"}
response.__enter__.return_value.read.return_value = b'{"projects":true}'
with patch("routes.canvas.urllib.request.urlopen", return_value=response) as urlopen:
assert canvas._handle_proxy(h, _parsed("path=/api/v1/capabilities")) is True
assert h.status == 200
assert h.body == b'{"projects":true}'
assert urlopen.call_args.args[0].full_url.endswith("/api/v1/capabilities")
def test_proxy_post_method_override_forwards_put_without_upstream_put_route():
h = _Handler(b'{"spec":{"schemaVersion":1}}')
h.command = "POST"
response = MagicMock()
response.__enter__.return_value.status = 200
response.__enter__.return_value.headers = {"Content-Type": "application/json"}
response.__enter__.return_value.read.return_value = b'{"id":"screen-1"}'
with patch("routes.canvas.urllib.request.urlopen", return_value=response) as urlopen:
assert canvas._handle_proxy(
h,
_parsed("path=/api/v1/projects/project-1/screens/screen-1&method=PUT"),
) is True
assert h.status == 200
req = urlopen.call_args.args[0]
assert req.full_url.endswith("/api/v1/projects/project-1/screens/screen-1")
assert req.get_method() == "PUT"
def test_proxy_allows_variant_and_prototype_contract_paths():
assert canvas._is_allowed("/api/v1/projects/project-1/variants")
assert canvas._is_allowed("/api/v1/projects/project-1/prototype-edges")
assert canvas._is_allowed("/api/v1/projects/project-1/export")
def test_status_includes_canva_editor_capabilities():
h = _Handler()
health = MagicMock()
health.__enter__.return_value.status = 200
health.__enter__.return_value.headers = {"Content-Type": "application/json"}
health.__enter__.return_value.read.return_value = b'{"status":"ok"}'
caps = MagicMock()
caps.__enter__.return_value.status = 200
caps.__enter__.return_value.headers = {"Content-Type": "application/json"}
caps.__enter__.return_value.read.return_value = b'{"projects":true,"variants":false}'
with patch("routes.canvas.urllib.request.urlopen", side_effect=[health, caps]):
assert canvas._handle_status(h, None) is True
payload = json.loads(h.body)
assert h.status == 200
assert payload["ok"] is True
assert payload["capabilities"] == {"projects": True, "variants": False}
def test_design_context_returns_seed_contract():
h = _Handler()
with patch("routes.canvas._load_secondbrain_hints", return_value={"status": "unconfigured", "hints": []}):
assert canvas._handle_design_context(h, _parsed("brand_id=planb")) is True
assert h.status == 200
payload = json.loads(h.body)
assert payload["brand_id"] == "planb"
assert payload["resolved_brand_id"] == "607740e8-8b76-4455-9c4f-eadbe9b4168c"
def test_design_context_reads_cmo_brand_profile_cache(tmp_path, monkeypatch):
db_path = tmp_path / "cmo.db"
conn = sqlite3.connect(db_path)
conn.execute(
"""CREATE TABLE brand_profile_cache (
brand_id TEXT PRIMARY KEY,
json TEXT,
version TEXT,
fetched_at TEXT
)"""
)
conn.execute(
"INSERT INTO brand_profile_cache (brand_id, json, version, fetched_at) VALUES (?, ?, ?, ?)",
(
"607740e8-8b76-4455-9c4f-eadbe9b4168c",
json.dumps({
"brand": {"name": "planb", "displayName": "Plan B Brand", "description": "Dark mode brand."},
"brand_guideline": "Use confident fr-CA food copy. Never imply industrial food.",
"design_md": "# DESIGN\nUse Montserrat.",
"palette": [
{"path": "color.primary", "type": "color", "valueJson": "{\"dark\":\"#FD5000\"}"},
],
"typography": [{"path": "font.family.body", "valueJson": "\"Montserrat\""}],
"spacing": [{"path": "spacing.4", "valueJson": "4"}],
"voice": {"dos": ["Be specific"], "donts": ["No scarcity"]},
"tokens_imported": True,
}),
"v1",
"2026-05-27T12:00:00Z",
),
)
conn.commit()
conn.close()
monkeypatch.setenv("CANVAS_CMO_DB", str(db_path))
h = _Handler()
with patch("routes.canvas._load_secondbrain_hints", return_value={
"status": "ready",
"hints": ["Secondbrain capsule 42: Use calm operator UI patterns."],
}):
assert canvas._handle_design_context(h, _parsed("brand_id=planb")) is True
payload = json.loads(h.body)
assert h.status == 200
assert payload["status"] == "ready"
assert payload["source"] == "cmo.brand_profile_cache"
assert payload["cache_brand_id"] == "607740e8-8b76-4455-9c4f-eadbe9b4168c"
assert payload["source_version"] == "v1"
assert payload["brand"]["displayName"] == "Plan B Brand"
assert payload["brand_guideline_text"].startswith("Use confident")
assert payload["palette_tokens"][0]["value"]["dark"] == "#FD5000"
assert "Be specific" in payload["voice_rules"]
assert any("BTE brand guideline" in hint for hint in payload["memory_hints"])
assert any("Secondbrain capsule 42" in hint for hint in payload["memory_hints"])
assert payload["secondbrain"]["status"] == "ready"
def test_design_context_reports_missing_cache_without_inventing_context(tmp_path, monkeypatch):
db_path = tmp_path / "cmo.db"
conn = sqlite3.connect(db_path)
conn.execute(
"""CREATE TABLE brand_profile_cache (
brand_id TEXT PRIMARY KEY,
json TEXT,
version TEXT,
fetched_at TEXT
)"""
)
conn.commit()
conn.close()
monkeypatch.setenv("CANVAS_CMO_DB", str(db_path))
h = _Handler()
assert canvas._handle_design_context(h, _parsed("brand_id=missing-brand")) is True
payload = json.loads(h.body)
assert payload["status"] == "missing"
assert payload["brand_guideline_text"] == ""
assert payload["memory_hints"] == []
assert "missing-brand" in payload["error"]
def test_events_returns_sse_seed():
h = _Handler()
assert canvas._handle_events(h, None) is True
assert h.status == 200
assert h.headers_out["Content-Type"].startswith("text/event-stream")
assert b"canvas.connected" in h.body
def test_command_generates_persists_and_replays_events():
body = json.dumps({"prompt": "Plan B dashboard", "variants": 2}).encode("utf-8")
h = _Handler(body)
h.command = "POST"
with patch("routes.canvas._ensure_project", return_value={"id": "project-1", "name": "Hermes Canvas"}), \
patch("routes.canvas._load_design_context", return_value={
"brand_id": "planb",
"resolved_brand_id": "607740e8-8b76-4455-9c4f-eadbe9b4168c",
"status": "ready",
"source": "cmo.brand_profile_cache",
"source_version": "v1",
"fetched_at": "2026-05-27T12:00:00Z",
"brand": {"displayName": "Plan B Brand"},
"memory_hints": ["Use BTE palette tokens: #FD5000"],
"design_md_excerpt": "Use Montserrat.",
"tokens_imported": True,
}), \
patch("routes.canvas._canva_json") as canva_json:
canva_json.side_effect = [
{"screenSpec": {"root": {"type": "text", "text": "A"}}},
{"id": "screen-1", "name": "Plan B dashboard", "spec": {"root": {"type": "text", "text": "A"}}},
{"screenSpec": {"root": {"type": "text", "text": "B"}}},
{"id": "screen-2", "name": "Plan B dashboard", "spec": {"root": {"type": "text", "text": "B"}}},
{"id": "variant-1"},
]
assert canvas._handle_command(h, None) is True
payload = json.loads(h.body)
assert h.status == 200
assert payload["ok"] is True
assert payload["design_context"]["status"] == "ready"
assert len(payload["screens"]) == 2
assert any(event["type"] == "canvas.screen.persisted" for event in payload["events"])
first_generate_payload = canva_json.call_args_list[0].args[2]
assert "BTE/CMO design constraints" in first_generate_payload["prompt"]
assert "#FD5000" in first_generate_payload["prompt"]
events = _Handler()
assert canvas._handle_events(events, _parsed("format=json&since=0")) is True
replay = json.loads(events.body)
assert replay["ok"] is True
assert any(event["type"] == "canvas.command.completed" for event in replay["events"])
def test_compose_generation_prompt_keeps_prompt_plain_when_context_missing():
assert canvas._compose_generation_prompt("Make a landing page", {"status": "missing"}) == "Make a landing page"
def test_secondbrain_hints_query_uses_local_psql(monkeypatch):
monkeypatch.delenv("SECONDBRAIN_DATABASE_URL", raising=False)
monkeypatch.setenv("SECONDBRAIN_DB_PASSWORD", "test-password")
completed = SimpleNamespace(
returncode=0,
stdout=json.dumps([
{
"id": 42,
"title": "Headless everything for personal AI",
"summary": "Keep the command center sovereign and web-native.",
"sector": "svrnty",
"created_at": "2026-05-27T12:00:00Z",
}
]),
stderr="",
)
with patch("routes.canvas.subprocess.run", return_value=completed) as run:
result = canvas._load_secondbrain_hints({"displayName": "Plan B"}, "brand-1")
assert result["status"] == "ready"
assert result["source"] == "secondbrain.knowledge_capsules"
assert result["capsules"][0]["id"] == 42
assert "command center" in result["hints"][0]
cmd = run.call_args.args[0]
assert cmd[0] == "psql"
assert "-tA" in cmd

View File

@ -0,0 +1,77 @@
import json
from io import BytesIO
from unittest.mock import patch
from routes import canvas
class _Handler:
command = "GET"
headers = {}
def __init__(self):
self.body = b""
self.status = None
self.sent_headers = {}
self.rfile = BytesIO()
@property
def wfile(self):
handler = self
class _Writer:
def write(self, data):
handler.body += data
return _Writer()
def send_response(self, status):
self.status = status
def send_header(self, key, value):
self.sent_headers[key] = value
def end_headers(self):
pass
def test_tools_manifest_exposes_cmo_canvas_surface():
handler = _Handler()
assert canvas._handle_tools(handler, None) is True
payload = json.loads(handler.body)
names = {tool["name"] for tool in payload["tools"]}
assert handler.status == 200
assert payload["tool_surface"] == "cmo-canvas-v0"
assert "canvas.generate_screen" in names
assert "canvas.command" in names
assert "canvas.generate_from_image" in names
assert "canvas.update_screen" in names
assert "canvas.record_variant" in names
assert "canvas.connect_prototype_edge" in names
assert "canvas.export_project" in names
assert "upstream-hermes-webui-edits" in payload["non_goals"]
def test_canvas_proxy_rejects_unlisted_paths():
handler = _Handler()
parsed = type("Parsed", (), {"query": "path=/api/admin/secrets"})()
assert canvas._handle_proxy(handler, parsed) is True
payload = json.loads(handler.body)
assert handler.status == 400
assert payload["ok"] is False
def test_canvas_status_reports_offline_service_without_failing_request():
handler = _Handler()
with patch("routes.canvas.urllib.request.urlopen", side_effect=OSError("offline")):
assert canvas._handle_status(handler, None) is True
payload = json.loads(handler.body)
assert handler.status == 200
assert payload["ok"] is False
assert payload["service"] == "canva-editor"

View File

@ -0,0 +1,39 @@
"""Static checks for the Canvas panel assets."""
from pathlib import Path
ROOT = Path(__file__).resolve().parents[2]
CANVAS_JS = ROOT / "static" / "canvas.js"
CANVAS_CSS = ROOT / "static" / "canvas.css"
def test_canvas_assets_exist():
assert CANVAS_JS.is_file()
assert CANVAS_CSS.is_file()
def test_canvas_js_is_idempotent_and_uses_plugin_routes():
src = CANVAS_JS.read_text()
assert "__svrntyCanvasLoaded" in src
assert "request: _request" in src
assert "svrnty:canvas-generate" in src
assert 'name: "canvas"' in src
assert "/api/canvas/status" in src
assert "/api/canvas/tools" in src
assert "/api/canvas/proxy" in src
assert "/api/canvas/command" in src
assert "/api/canvas/events" in src
assert "/api/canvas/design-context" in src
assert "designContext" in src
assert "BTE context ready" in src
assert "data-canvas-action=\"save\"" in src
assert "data-canvas-action=\"prototype-link\"" in src
assert "/prototype-edges" in src
assert "/export" in src
assert "status.capabilities" in src
def test_canvas_css_is_panel_scoped():
src = CANVAS_CSS.read_text()
assert ".svrnty-canvas-" in src
assert ".svrnty-canvas-overlay" in src

View File

@ -0,0 +1,51 @@
from unittest.mock import Mock
from routes import cortex_os_extension
def test_adapter_constants_preserve_s27_identity():
assert cortex_os_extension.PACKAGE_ID == "cortex-os-extension"
assert cortex_os_extension.MEMBER_ID == "s23-runtime-health-read-only-slice"
assert cortex_os_extension.ACTIVE_DEVELOPMENT_MOUNT_TARGET == "hermes-webui"
assert cortex_os_extension.RUNTIME_HEALTH_METHOD == "GET"
assert cortex_os_extension.RUNTIME_HEALTH_ROUTE == "/api/cortex-os/runtime-health"
def test_adapter_registers_runtime_health_route_and_assets_only():
api = Mock()
api.logger.return_value = Mock()
cortex_os_extension.register(api, "svrnty")
api.inject_stylesheet.assert_called_once_with(
"/plugins/svrnty/cortex-os/runtime-health/runtime_health.css"
)
api.inject_script.assert_called_once_with(
"/plugins/svrnty/cortex-os/runtime-health/runtime_health.js"
)
api.register_route.assert_called_once()
method, path = api.register_route.call_args.args[:2]
assert method == "/api/cortex-os/runtime-health"
assert path == "GET"
api.register_static.assert_not_called()
api.config_get.assert_not_called()
api.register_audio_attachment_processor.assert_not_called()
def test_adapter_source_has_no_forbidden_runtime_discovery():
source = cortex_os_extension.__loader__.get_source(cortex_os_extension.__name__)
forbidden = [
"glob(",
"rglob(",
"subprocess",
"os.walk",
"sot/08-OUTPUTS",
"hermes_webui",
"hermes_agent",
"register_static",
"config_get",
"register_audio_attachment_processor",
"product readiness",
]
for snippet in forbidden:
assert snippet not in source

View File

@ -0,0 +1,120 @@
import io
import json
from types import SimpleNamespace
from unittest.mock import Mock
from routes import cortex_os_runtime_health as route
class FakeHandler:
def __init__(self, command="GET"):
self.command = command
self.status_code = None
self.headers = []
self.wfile = io.BytesIO()
def send_response(self, status_code):
self.status_code = status_code
def send_header(self, name, value):
self.headers.append((name, value))
def end_headers(self):
return None
def payload(self):
return json.loads(self.wfile.getvalue().decode("utf-8"))
def test_register_wires_get_runtime_health_route():
api = Mock()
api.logger.return_value = Mock()
route.register(api)
api.register_route.assert_called_once_with(
"/api/cortex-os/runtime-health",
"GET",
route._handle_runtime_health,
)
def test_get_returns_runtime_health_envelope():
handler = FakeHandler()
assert route._handle_runtime_health(handler, SimpleNamespace(query="")) is True
payload = handler.payload()
assert handler.status_code == 200
assert payload["ok"] is True
assert payload["result"]["schema_version"] == "0.1.0"
assert payload["result"]["cortex_os_contract_id"] == "runtime-health/v0.1"
assert payload["result"]["host_runtime_id"] == "webui"
assert payload["result"]["host_adapter_id"] == "hermes"
assert payload["result"]["host_adapter_kind"] == "development_host"
assert payload["result"]["checked_at"] == "2026-05-29T00:00:00Z"
assert payload["result"]["status"] == "unknown"
assert payload["result"]["readiness"] == "not_configured"
assert "display_summary" in payload["result"]
assert {signal["signal_id"] for signal in payload["result"]["signals"]} == {
"health",
"agent_health",
"dashboard_status",
}
assert payload["result"]["authority"]["runtime_state_mutation"] is False
assert all(value is False for value in payload["result"]["authority"].values())
def test_runtime_health_envelope_rejects_old_shape_fields():
result = route.runtime_health_payload()
forbidden_top_level = {"contract_id", "summary"}
forbidden_authority = {
"read_only",
"mcp_exposure",
"profile_exposure_change",
"delegated_memory_grant",
}
assert forbidden_top_level.isdisjoint(result)
assert forbidden_authority.isdisjoint(result["authority"])
assert isinstance(result["redactions"][0], dict)
assert isinstance(result["warnings"][0], dict)
assert isinstance(result["source_trace"], list)
for signal in result["signals"]:
assert {"signal_id", "state", "summary", "redacted"} == set(signal)
def test_host_signal_mapping_is_closed_and_redacted():
payload = route.runtime_health_payload(
{
"health": "healthy",
"agent_health": "degraded",
"dashboard_status": "unavailable",
"health_detail": "http://localhost:8000/raw",
"agent_health_detail": "/home/svrnty/private",
"dashboard_status_detail": "token=abc123",
}
)
assert payload["status"] == "unavailable"
assert {signal["summary"] for signal in payload["signals"]} == {"redacted"}
assert {signal["redacted"] for signal in payload["signals"]} == {True}
assert [entry["source_id"] for entry in payload["source_trace"]] == [
"runtime-health-contract",
"hermes-host-adapter-contract",
"hermes-runtime-health-slice",
]
def test_non_get_and_query_targets_are_rejected():
post_handler = FakeHandler(command="POST")
query_handler = FakeHandler()
route._handle_runtime_health(post_handler, SimpleNamespace(query=""))
route._handle_runtime_health(query_handler, SimpleNamespace(query="target=raw"))
assert post_handler.status_code == 405
assert post_handler.payload()["ok"] is False
assert query_handler.status_code == 400
assert query_handler.payload()["error"]["code"] == "query_not_allowed"

View File

@ -0,0 +1,138 @@
from pathlib import Path
ROOT = Path(__file__).resolve().parents[2]
JS = ROOT / "static" / "cortex-os" / "runtime-health" / "runtime_health.js"
CSS = ROOT / "static" / "cortex-os" / "runtime-health" / "runtime_health.css"
def test_runtime_health_assets_exist_and_render_all_states():
js = JS.read_text(encoding="utf-8")
css = CSS.read_text(encoding="utf-8")
assert "/api/cortex-os/runtime-health" in js
for state in ["healthy", "degraded", "unavailable", "unknown", "loading", "error", "redacted"]:
assert state in js
assert state in css
def test_runtime_health_display_has_no_hidden_write_surface():
js = JS.read_text(encoding="utf-8")
forbidden = [
"localStorage",
"sessionStorage",
"document.cookie",
"indexedDB",
"caches",
"navigator.sendBeacon",
"WebSocket",
"EventSource",
"BroadcastChannel",
"SharedWorker",
"ServiceWorker",
"window.open",
"eval(",
"new Function",
"JSON.stringify",
"Object.keys(payload",
"Object.entries(payload",
"clipboard",
"innerHTML",
"insertAdjacentHTML",
"outerHTML",
"DOMParser",
"template.innerHTML",
"window.location",
"createElement(\"script\"",
"<form",
".submit(",
"POST",
"PUT",
"PATCH",
"DELETE",
"/api/gateway/status",
"/api/tools",
"/api/mcp",
]
for snippet in forbidden:
assert snippet not in js
def test_runtime_health_css_is_scoped_to_panel():
css = CSS.read_text(encoding="utf-8")
assert ".cortex-os-runtime-health" in css
assert "body {" not in css
assert "#app" not in css
def test_runtime_health_mounts_only_in_cortex_os_panel_context():
js = JS.read_text(encoding="utf-8")
assert 'window.addEventListener("svrnty:panel-switch", handlePanelSwitch)' in js
assert 'detail.name === "cortex-os"' in js
assert 'document.querySelector("main.main")' in js
assert "document.body" not in js
assert 'document.addEventListener("DOMContentLoaded", loadRuntimeHealth' not in js
assert "loadRuntimeHealth();" in js
assert js.count("/api/cortex-os/runtime-health") == 1
def test_runtime_health_display_consumes_canonical_envelope_only():
js = JS.read_text(encoding="utf-8")
for snippet in [
"payload.result.display_summary",
"displaySummary.headline",
"displaySummary.detail",
"signal.signal_id",
"signal.state",
"signal.summary",
"payload.result.warnings",
"payload.result.redactions",
"payload.result.source_trace",
"payload.result.authority",
]:
assert snippet in js
for forbidden in [
"payload.result.summary",
"signal.name",
"signal.status",
"signal.detail",
"contract_id",
"source_trace.live_probe",
"source_trace.raw_payload_passthrough",
"authority.read_only",
"authority.mcp_exposure",
"authority.profile_exposure_change",
"authority.delegated_memory_grant",
]:
assert forbidden not in js
def test_runtime_health_detail_surface_is_bounded_and_text_only():
js = JS.read_text(encoding="utf-8")
css = CSS.read_text(encoding="utf-8")
for snippet in [
'details.setAttribute("data-role", "details")',
"renderDetails(",
"renderKeyValues(",
"signals.slice(0, 3)",
"warnings.slice(0, 3)",
"redactions.slice(0, 4)",
"sourceTrace.slice(0, 3)",
"allAuthorityDenied",
".every(function (key)",
"textContent",
]:
assert snippet in js
for snippet in [
".cortex-os-runtime-health__details",
".cortex-os-runtime-health__detail-group",
".cortex-os-runtime-health__detail-list",
]:
assert snippet in css
for forbidden in ["body {", "main {", "#app", ".panel"]:
assert forbidden not in css

View File

@ -16,3 +16,18 @@ def test_svrnty_tabs_participate_in_panel_switching():
src = NAV_JS.read_text()
assert "const TABS = [" in src
assert "const OUR_IDS = TABS.map((t) => t.id)" in src
assert 'id: "canvas"' in src
def test_canvas_tab_is_registered():
src = NAV_JS.read_text()
assert '{ id: "canvas"' in src
assert "svrnty-showing-\" + id" in src
def test_cortex_os_tab_is_registered_once():
src = NAV_JS.read_text()
assert src.count('id: "cortex-os"') == 1
assert '"cortex-os":' in src
for tab_id in ['id: "adwright"', 'id: "bte"', 'id: "canvas"']:
assert tab_id in src