diff --git a/CONNECTION-MAP.md b/CONNECTION-MAP.md index 1b2b93e..4f0c0b8 100644 --- a/CONNECTION-MAP.md +++ b/CONNECTION-MAP.md @@ -2,7 +2,7 @@ **Upstream version:** v0.51.118 **Plugin version:** 0.5.0 -**Total dependencies:** 39 (29 public API · 0 forced internal · 10 frontend) +**Total dependencies:** 61 (40 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 @@ -23,16 +23,27 @@ | `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:51` | `api.inject_stylesheet` | `api.inject_stylesheet(f"/plugins/{STATIC_PREFIX}/cortex-os/runtime-health/runtime_health.c` | -| `plugin.py:52` | `api.inject_script` | `api.inject_script(f"/plugins/{STATIC_PREFIX}/cortex-os/runtime-health/runtime_health.js")` | +| `plugin.py:48` | `api.inject_stylesheet` | `api.inject_stylesheet(f"/plugins/{STATIC_PREFIX}/canvas.css")` | +| `plugin.py:49` | `api.inject_script` | `api.inject_script(f"/plugins/{STATIC_PREFIX}/canvas.js")` | +| `plugin.py:51` | `api.inject_stylesheet` | `api.inject_stylesheet(f"/plugins/{STATIC_PREFIX}/umbrella_inline.css")` | +| `plugin.py:52` | `api.inject_script` | `api.inject_script(f"/plugins/{STATIC_PREFIX}/umbrella_inline.js")` | +| `plugin.py:54` | `api.inject_stylesheet` | `api.inject_stylesheet(f"/plugins/{STATIC_PREFIX}/cortex-os/runtime-health/runtime_health.c` | +| `plugin.py:55` | `api.inject_script` | `api.inject_script(f"/plugins/{STATIC_PREFIX}/cortex-os/runtime-health/runtime_health.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_runtime_health.py:26` | `api.logger` | `log = api.logger("svrnty.routes.cortex_os_runtime_health")` | | `routes/cortex_os_runtime_health.py:27` | `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")` | @@ -58,6 +69,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/` | diff --git a/manifest.yaml b/manifest.yaml index 6434247..b535d4b 100644 --- a/manifest.yaml +++ b/manifest.yaml @@ -33,11 +33,13 @@ assets: - /plugins/svrnty/svrnty_nav.js - /plugins/svrnty/adwright.js - /plugins/svrnty/bte.js + - /plugins/svrnty/canvas.js - /plugins/svrnty/umbrella_inline.js stylesheets: - /plugins/svrnty/app.css - /plugins/svrnty/adwright.css - /plugins/svrnty/bte.css + - /plugins/svrnty/canvas.css - /plugins/svrnty/umbrella_inline.css # Routes this plugin registers at load time (declarative cross-check vs runtime). @@ -49,6 +51,14 @@ 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 } # Audio-attachment processors (called by streaming.py before agent receives message). audio_processors: diff --git a/plugin.py b/plugin.py index 0da9674..3360316 100644 --- a/plugin.py +++ b/plugin.py @@ -44,6 +44,9 @@ 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") @@ -79,6 +82,7 @@ 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) ✓ "cortex_os_runtime_health", # S23.0-I4 — Cortex OS Runtime Health read-only slice ✓ ] diff --git a/routes/canvas.py b/routes/canvas.py new file mode 100644 index 0000000..75b65a0 --- /dev/null +++ b/routes/canvas.py @@ -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 diff --git a/scripts/boot-smoke.py b/scripts/boot-smoke.py index eaa14d4..2eedd1d 100755 --- a/scripts/boot-smoke.py +++ b/scripts/boot-smoke.py @@ -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: diff --git a/scripts/canvas-visual-smoke.py b/scripts/canvas-visual-smoke.py new file mode 100755 index 0000000..8491b7f --- /dev/null +++ b/scripts/canvas-visual-smoke.py @@ -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()) diff --git a/scripts/cmo-natural-canvas-smoke.py b/scripts/cmo-natural-canvas-smoke.py new file mode 100755 index 0000000..ba9ff8b --- /dev/null +++ b/scripts/cmo-natural-canvas-smoke.py @@ -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()) diff --git a/static/canvas.css b/static/canvas.css new file mode 100644 index 0000000..0dfd847 --- /dev/null +++ b/static/canvas.css @@ -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; + } +} diff --git a/static/canvas.html b/static/canvas.html new file mode 100644 index 0000000..e958cbd --- /dev/null +++ b/static/canvas.html @@ -0,0 +1,39 @@ + + + + + + Sovereign Canvas + + + +
+
+
Sovereign Canvas
+
Checking...
+
+
+
+
Open Hermes WebUI and select Canvas for the live command-center panel.
+
+ +
+
+ + + diff --git a/static/canvas.js b/static/canvas.js new file mode 100644 index 0000000..b4b6211 --- /dev/null +++ b/static/canvas.js @@ -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 "); + 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 = + '
' + + '
Sovereign Canvas
' + + '
Checking...
' + + '' + + '' + + '' + + '' + + '' + + '
' + + '
' + + '
' + + '' + + '
'; + 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 = '
Prompt CMO to generate a screen. Artboards, variants, edits, prototype links, and exports will accumulate here.
'; + return; + } + target.innerHTML = state.artboards.map((board, idx) => + '
' + + '
' + _esc(board.title) + '' + _esc(board.dirty ? "edited" : board.badge) + '
' + + '
' + + '' + + '' + + '' + + '' + + '
' + + '
' + _renderSpec(board.spec) + '
' + + '
' + ).join(""); + } + + function _renderSpec(spec) { + if (!spec) return '
Empty spec
'; + 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 '
' + _esc(text) + childHtml + '
'; + } + + 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) => + '
' + _esc(tool.name) + '
' + _esc(tool.purpose || "") + '
' + ).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) => + '
' + _esc(row.at) + '
' + _esc(row.message) + '
' + ).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) => ({ + "<": "<", + ">": ">", + "&": "&", + '"': """, + }[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(); + } +})(); diff --git a/static/svrnty_nav.js b/static/svrnty_nav.js index 090d322..3570bfb 100644 --- a/static/svrnty_nav.js +++ b/static/svrnty_nav.js @@ -22,10 +22,13 @@ '', bte: '', + canvas: + '', }; 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" }, ]; function _svg(iconPath, size, stroke) { diff --git a/tests/evals/test_features.py b/tests/evals/test_features.py index 898e25f..a0fb601 100644 --- a/tests/evals/test_features.py +++ b/tests/evals/test_features.py @@ -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(): diff --git a/tests/integration/test_loader_contract.py b/tests/integration/test_loader_contract.py index 0bcdb25..ba9f776 100644 --- a/tests/integration/test_loader_contract.py +++ b/tests/integration/test_loader_contract.py @@ -60,6 +60,14 @@ 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 @@ -67,6 +75,8 @@ def test_loader_register_wires_our_plugin(loader, monkeypatch): 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 diff --git a/tests/unit/test_canvas.py b/tests/unit/test_canvas.py new file mode 100644 index 0000000..06673a7 --- /dev/null +++ b/tests/unit/test_canvas.py @@ -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 diff --git a/tests/unit/test_canvas_route.py b/tests/unit/test_canvas_route.py new file mode 100644 index 0000000..92f6163 --- /dev/null +++ b/tests/unit/test_canvas_route.py @@ -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" diff --git a/tests/unit/test_canvas_static.py b/tests/unit/test_canvas_static.py new file mode 100644 index 0000000..7994301 --- /dev/null +++ b/tests/unit/test_canvas_static.py @@ -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 diff --git a/tests/unit/test_svrnty_nav_js.py b/tests/unit/test_svrnty_nav_js.py index 384b895..b3153f9 100644 --- a/tests/unit/test_svrnty_nav_js.py +++ b/tests/unit/test_svrnty_nav_js.py @@ -16,3 +16,10 @@ 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