from __future__ import annotations import html import importlib.util import os import sys from pathlib import Path from typing import Any from card_store import ( NANOBOT_WORKSPACE, json_script_text, normalize_card_id, persist_card, template_html_path, ) from session_store import normalize_session_chat_id BASE_DIR = Path(__file__).resolve().parent NANOBOT_REPO_DIR = BASE_DIR.parent / "nanobot" NANOBOT_RUNTIME_WORKSPACE = Path( os.getenv( "NANOBOT_RUNTIME_WORKSPACE", str( (NANOBOT_WORKSPACE / "workspace") if (NANOBOT_WORKSPACE / "workspace").exists() else NANOBOT_WORKSPACE ), ) ).expanduser() WORKBENCH_DIR = NANOBOT_RUNTIME_WORKSPACE / "workbench" def ensure_workbench_store_dirs() -> None: WORKBENCH_DIR.mkdir(parents=True, exist_ok=True) _script_modules: dict[str, Any] = {} def _load_script_module(module_name: str, script_path: Path): cached = _script_modules.get(module_name) if cached is not None: return cached if not script_path.exists(): raise RuntimeError(f"missing helper script: {script_path}") spec = importlib.util.spec_from_file_location(module_name, script_path) if spec is None or spec.loader is None: raise RuntimeError(f"failed to load helper script: {script_path}") module = importlib.util.module_from_spec(spec) sys.modules.setdefault(module_name, module) spec.loader.exec_module(module) _script_modules[module_name] = module return module def _workbench_board_module(): ensure_workbench_store_dirs() module = _load_script_module( "nanobot_web_workbench_board", NANOBOT_REPO_DIR / "scripts" / "workbench_board.py", ) module.ensure_workbench(WORKBENCH_DIR) return module def _coerce_workbench_record(raw: dict[str, Any]) -> dict[str, Any] | None: item_id = normalize_card_id(str(raw.get("id", ""))) chat_id = normalize_session_chat_id(str(raw.get("chat_id", "web") or "web")) if not item_id or not chat_id: return None kind = str(raw.get("kind", "text") or "text").strip().lower() if kind not in {"text", "question"}: kind = "text" raw_choices = raw.get("choices", []) choices = [str(choice) for choice in raw_choices] if isinstance(raw_choices, list) else [] raw_template_state = raw.get("template_state", {}) template_state = raw_template_state if isinstance(raw_template_state, dict) else {} return { "id": item_id, "chat_id": chat_id, "kind": kind, "title": str(raw.get("title", "")), "content": str(raw.get("content", "")), "question": str(raw.get("question", "")), "choices": choices, "response_value": str(raw.get("response_value", "")), "slot": str(raw.get("slot", "")), "template_key": str(raw.get("template_key", "")), "template_state": template_state, "context_summary": str(raw.get("context_summary", "")), "promotable": bool(raw.get("promotable", True)), "source_card_id": str(raw.get("source_card_id", "")), "created_at": str(raw.get("created_at", "")), "updated_at": str(raw.get("updated_at", "")), } def _materialize_workbench_content(item: dict[str, Any]) -> str: if item.get("kind") != "text": return str(item.get("content", "")) template_key = str(item.get("template_key", "")).strip() if not template_key: return str(item.get("content", "")) html_path = template_html_path(template_key) try: template_html = html_path.read_text(encoding="utf-8") except Exception: return ( '
' f"Missing template: {html.escape(template_key)}" "
" ) state_payload = item.get("template_state", {}) if not isinstance(state_payload, dict): state_payload = {} item_id = html.escape(str(item.get("id", ""))) safe_template_key = html.escape(template_key) return ( f'
' f'' f"{template_html}" "
" ) def _workbench_notification_target(raw: dict[str, Any]) -> bool: for key in ("surface", "target", "placement"): value = str(raw.get(key, "") or "").strip().lower() if value == "workbench": return True return False def _persist_workbench_item( raw: dict[str, Any], *, default_chat_id: str | None = None ) -> dict[str, Any] | None: normalized_chat_id = normalize_session_chat_id( str(raw.get("chat_id", default_chat_id or "web") or default_chat_id or "web") ) if not normalized_chat_id: return None template_state = raw.get("template_state", {}) if not isinstance(template_state, dict): template_state = {} workbench_board = _workbench_board_module() item = workbench_board.upsert_item( WORKBENCH_DIR, chat_id=normalized_chat_id, item_id=str(raw.get("id", "") or ""), kind=str(raw.get("kind", "text") or "text"), title=str(raw.get("title", "")), content=str(raw.get("content", "")), question=str(raw.get("question", "")), choices=[str(choice) for choice in raw.get("choices", [])] if isinstance(raw.get("choices"), list) else [], response_value=str(raw.get("response_value", "")), slot=str(raw.get("slot", "")), template_key=str(raw.get("template_key", "")), template_state=template_state, context_summary=str(raw.get("context_summary", "")), promotable=bool(raw.get("promotable", True)), source_card_id=str(raw.get("source_card_id", "")), ) normalized = _coerce_workbench_record(item) if normalized is None: return None normalized["content"] = _materialize_workbench_content(normalized) return normalized def _load_workbench_items(chat_id: str) -> list[dict[str, Any]]: workbench_board = _workbench_board_module() items = workbench_board.collect_items(WORKBENCH_DIR, chat_id) rendered: list[dict[str, Any]] = [] for item in items: normalized = _coerce_workbench_record(item) if normalized is None: continue normalized["content"] = _materialize_workbench_content(normalized) rendered.append(normalized) rendered.sort(key=lambda item: str(item.get("updated_at", "")), reverse=True) return rendered def _delete_workbench_item(chat_id: str, item_id: str) -> bool: workbench_board = _workbench_board_module() return bool(workbench_board.delete_item(WORKBENCH_DIR, chat_id, item_id)) def _promote_workbench_item(chat_id: str, item_id: str) -> dict[str, Any]: workbench_board = _workbench_board_module() item = workbench_board.load_item(WORKBENCH_DIR, chat_id, item_id) card_payload = { "id": f"workbench-{item['id']}", "kind": item.get("kind", "text"), "title": item.get("title", ""), "content": item.get("content", ""), "question": item.get("question", ""), "choices": item.get("choices", []), "response_value": item.get("response_value", ""), "slot": f"workbench:{chat_id}:{item['id']}", "lane": "context", "priority": 78, "state": "active", "template_key": item.get("template_key", ""), "template_state": item.get("template_state", {}), "context_summary": item.get("context_summary", ""), "chat_id": chat_id, } persisted = persist_card(card_payload) if persisted is None: raise RuntimeError("failed to promote workbench item to card") workbench_board.delete_item(WORKBENCH_DIR, chat_id, item_id) return persisted coerce_workbench_record = _coerce_workbench_record persist_workbench_item = _persist_workbench_item load_workbench_items = _load_workbench_items delete_workbench_item = _delete_workbench_item promote_workbench_item = _promote_workbench_item workbench_notification_target = _workbench_notification_target __all__ = [ "WORKBENCH_DIR", "coerce_workbench_record", "delete_workbench_item", "ensure_workbench_store_dirs", "load_workbench_items", "persist_workbench_item", "promote_workbench_item", "workbench_notification_target", ]