246 lines
8.4 KiB
Python
246 lines
8.4 KiB
Python
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 (
|
|
'<div style="padding:16px;border:1px solid #fecaca;border-radius:12px;'
|
|
'background:#fef2f2;color:#991b1b;font:600 14px/1.4 system-ui,sans-serif;">'
|
|
f"Missing template: {html.escape(template_key)}"
|
|
"</div>"
|
|
)
|
|
|
|
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'<div data-nanobot-card-root data-card-id="{item_id}" data-template-key="{safe_template_key}">'
|
|
f'<script type="application/json" data-card-state>{json_script_text(state_payload)}</script>'
|
|
f"{template_html}"
|
|
"</div>"
|
|
)
|
|
|
|
|
|
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",
|
|
]
|