#!/usr/bin/env python3
"""
Hermes 网页聊天后端 v4 — JSON-RPC 协议版
------------------------------------------------
架构:
浏览器 → ws://host/ws/chat → 本服务 → ws://127.0.0.1:9119/api/ws → Hermes TUI Gateway
↓
JSON-RPC 协议
- session.create
- prompt.submit
- 事件流: message.delta, reasoning.delta,
thinking.delta, tool.start, tool.complete,
message.complete
API:
WS: /ws/chat WebSocket 聊天
HTTP: /chat/tasks 任务列表
/chat/tree wiki 知识树
/chat/history 聊天记录
/chat/prd PRD 文档
v4 — 改用 JSON-RPC 协议直连 Hermes Dashboard，支持推理面板 + 流式渲染
"""
HERMES_DASHBOARD_WS = "ws://127.0.0.1:9119/api/ws"
HERMES_DASHBOARD_AUTH_TOKEN = ""
import asyncio
import json
import os
import sys
import sqlite3
import signal
import urllib.parse
import re
import time as _time_module
from datetime import datetime, timezone
import websockets
# ── 从 Dashboard HTML 中自动提取 session token ──
def _fetch_dashboard_token():
"""从 Dashboard 首页 HTML 中提取 __HERMES_SESSION_TOKEN__"""
import urllib.request
try:
resp = urllib.request.urlopen("http://127.0.0.1:9119/", timeout=3)
html = resp.read().decode('utf-8', errors='replace')
m = re.search(r'__HERMES_SESSION_TOKEN__="([^"]+)"', html)
if m:
return m.group(1)
except Exception as e:
print(f"[TOKEN FETCH] {e}", file=sys.stderr)
return ""
HOST = "0.0.0.0"
WS_PORT = 8765
HTTP_PORT = 8766
HERMES_HOME = os.path.expanduser('~/.hermes')
STATE_DB = os.path.join(HERMES_HOME, 'state.db')
# ── 数据库 ──
CONTENT_DB = "/root/content.db"
CHAT_HISTORY_DB = os.path.join(os.path.dirname(__file__), 'chat_history.db')
# ── 子Chat配置 ──
SOURCE_CONFIG = {
'chat': {'soul': None},
'chat-mian': {
'soul': os.path.join(os.path.dirname(__file__), 'soul_mian.md'),
'allowed_skills': [],
'blocked_skills': ['bt-search', 'bt_search'],
'allowed_toolsets': [],
'knowledge_root': '/root/知识库/阿棉/',
},
'chat-liangliang': {
'soul': os.path.join(os.path.dirname(__file__), 'soul_liangliang.md'),
'allowed_skills': [],
'blocked_skills': ['bt-search', 'bt_search'],
'allowed_toolsets': [],
'knowledge_root': '/root/知识库/亮亮/',
},
}
# ── 消息队列调度器 ──
chat_queues = {}
chat_queue_busy = {}
chat_queue_msg_id_counter = 0
# ── 任务看板 ──
task_board = {}
task_board_order = []
# ── WebSocket 连接状态 ──
active_connections = {} # ws -> {device_id, current_task_id}
# ── Task → Session 映射（持久化到 tasks 表） ──
# 同一个 task 复用同一个 kernel session_id，让内核保持上下文
# session_id 存在 content.db 的 tasks 表中，重启不丢失
# ═══════════════════════════════════════════
# 数据库操作
# ═══════════════════════════════════════════
def save_chat_message(task_id, role, content):
source = f"task-{task_id}"
now = datetime.now(timezone.utc).timestamp()
try:
conn = sqlite3.connect(CHAT_HISTORY_DB)
conn.execute(
"INSERT INTO messages (source, role, content, created_at) VALUES (?, ?, ?, ?)",
(source, role, content, now)
)
conn.commit()
conn.close()
except Exception:
pass
# 同步更新 task 更新时间为最新消息时间
_touch_task(task_id, now)
def _touch_task(task_id, ts=None):
"""更新 task 的 updated_at 时间"""
try:
now = datetime.fromtimestamp(ts, tz=timezone.utc).strftime('%Y-%m-%dT%H:%M:%S') if ts else datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%S')
conn = sqlite3.connect(CONTENT_DB)
conn.execute("UPDATE tasks SET updated_at=? WHERE id=?", (now, task_id))
conn.commit()
conn.close()
except Exception:
pass
def touch_task_api(task_id):
"""外部（微信/CLI）调用此函数更新 task 时间"""
_touch_task(task_id)
return True
def get_task_session_id(task_id):
"""从 tasks 表读取持久化的 session_id（内核存储用）"""
try:
conn = sqlite3.connect(CONTENT_DB)
row = conn.execute("SELECT session_id FROM tasks WHERE id=?", (task_id,)).fetchone()
conn.close()
return row[0] if row and row[0] else None
except Exception:
return None
def get_task_routing_session_id(task_id):
"""从 tasks 表读取 routing_session_id（网关路由用）"""
try:
conn = sqlite3.connect(CONTENT_DB)
row = conn.execute("SELECT routing_session_id FROM tasks WHERE id=?", (task_id,)).fetchone()
conn.close()
return row[0] if row and row[0] else None
except Exception:
return None
def resolve_task_for_source(source):
"""从 source（如 'chat'）找到最活跃的 task_id 用于历史查询"""
try:
conn = sqlite3.connect(CONTENT_DB)
rows = conn.execute(
"SELECT id, session_id FROM tasks WHERE status IN ('active','running') "
"ORDER BY updated_at DESC LIMIT 1"
).fetchall()
conn.close()
if rows:
return rows[0][0]
return None
except Exception:
return None
def _ensure_valid_session_id(task_id):
"""检查 session_id 是否有效。
注意：不再因 messages 为空就删 session。session 刚创建时 messages 可能为 0，
删掉会导致丢消息（下次请求创建新session，旧消息成孤儿）。
仅当 session_id 在 state.db 的 sessions 表中完全不存在时才视为失效。
"""
try:
sid = get_task_session_id(task_id)
if not sid:
return None
state_db = os.path.expanduser('~/.hermes/state.db')
conn = sqlite3.connect(state_db)
# 检查 sessions 表是否存在该 session（比查 messages 更准确）
session_exists = conn.execute(
"SELECT 1 FROM sessions WHERE id=?", (sid,)
).fetchone()
conn.close()
if not session_exists:
# session 在 state.db 中完全不存在（可能因为内核重置/迁移）
conn2 = sqlite3.connect(CONTENT_DB)
conn2.execute("UPDATE tasks SET session_id=NULL WHERE id=?", (task_id,))
conn2.commit()
conn2.close()
print(f"[HISTORY] stale session_id '{sid}' cleared for task {task_id} (session not in state.db)", file=sys.stderr)
return None
return sid
except Exception:
return None
def save_task_session_id(task_id, session_id, routing_session_id=''):
"""持久化 task → session_id 映射到 tasks 表"""
try:
conn = sqlite3.connect(CONTENT_DB)
conn.execute("UPDATE tasks SET session_id=?, routing_session_id=? WHERE id=?",
(session_id, routing_session_id, task_id))
conn.commit()
conn.close()
return True
except Exception:
return False
def load_chat_history_from_kernel(task_id, offset=0, limit=20):
"""从 Hermes 内核 state.db 拉聊天历史，不再用 chat_history.db"""
try:
# 1. 取 task 对应的 kernel session_id（自动校验失效 session）
session_id = _ensure_valid_session_id(task_id) or get_task_session_id(task_id)
if not session_id:
# 无 session → 降级到 chat_history.db
return load_chat_history_by_source(f"task-{task_id}", offset, limit)
# 2. 如有必要，迁移 chat_history.db 旧消息到 state.db
try:
ch_db = CHAT_HISTORY_DB
state_db = os.path.expanduser('~/.hermes/state.db')
if os.path.exists(ch_db):
ch_conn = sqlite3.connect(ch_db)
ch_rows = ch_conn.execute(
"SELECT role, content, created_at FROM messages WHERE source=? ORDER BY id",
(f"task-{task_id}",)
).fetchall()
ch_conn.close()
if ch_rows:
st_conn = sqlite3.connect(state_db)
count_before = st_conn.execute(
"SELECT COUNT(*) FROM messages WHERE session_id=?", (session_id,)
).fetchone()[0]
st_conn.execute("BEGIN")
for row in ch_rows:
ts = float(row[2]) if isinstance(row[2], (int, float)) else 0
# 用 WHERE NOT EXISTS 防止重复插入
st_conn.execute(
"INSERT INTO messages (session_id, role, content, timestamp, active) "
"SELECT ?, ?, ?, ?, 1 "
"WHERE NOT EXISTS ("
" SELECT 1 FROM messages WHERE session_id=? AND role=? AND content=? AND timestamp=?"
")",
(session_id, row[0], row[1] or '', ts,
session_id, row[0], row[1] or '', ts)
)
st_conn.execute(
"UPDATE sessions SET message_count = (SELECT COUNT(*) FROM messages WHERE session_id=?) WHERE id=?",
(session_id, session_id)
)
st_conn.commit()
count_after = st_conn.execute(
"SELECT COUNT(*) FROM messages WHERE session_id=?", (session_id,)
).fetchone()[0]
st_conn.close()
if count_after > count_before:
print(f"[HISTORY] Migrated {count_after - count_before} new msgs from chat_history.db to session {session_id}", file=sys.stderr)
except Exception as e:
print(f"[HISTORY] migration check error: {e}", file=sys.stderr)
# 3. 从 state.db 的 messages 表查
state_db = os.path.expanduser('~/.hermes/state.db')
conn = sqlite3.connect(state_db)
conn.row_factory = sqlite3.Row
max_limit = min(limit, 50)
rows = conn.execute(
"SELECT id, role, content, timestamp FROM messages "
"WHERE session_id=? AND role IN ('user', 'assistant') AND (content IS NOT NULL AND content != '') "
"ORDER BY id DESC LIMIT ? OFFSET ?",
(session_id, max_limit + 1, offset)
).fetchall()
conn.close()
has_more = len(rows) > max_limit
msgs = []
for r in rows[:max_limit]:
ts = r['timestamp']
if isinstance(ts, (int, float)):
time_str = datetime.fromtimestamp(ts).strftime('%m-%d %H:%M')
else:
time_str = str(ts)[:16] if ts else ''
content = r['content'] or ''
if len(content) > 3000:
content = content[:3000] + '…'
msgs.append({
'id': r['id'],
'role': r['role'],
'content': content,
'created_at': time_str,
})
msgs.reverse()
return {'messages': msgs, 'has_more': has_more}
except Exception as e:
print(f"[HISTORY] kernel query error: {e}", file=sys.stderr)
return {'messages': [], 'has_more': False, 'error': str(e)}
def load_chat_history(task_id, offset=0, limit=20):
source = f"task-{task_id}"
max_limit = min(limit, 50)
try:
conn = sqlite3.connect(CHAT_HISTORY_DB)
conn.row_factory = sqlite3.Row
rows = conn.execute(
"SELECT id, role, content, created_at FROM messages "
"WHERE source=? AND role IN ('user', 'assistant') "
"ORDER BY id DESC LIMIT ? OFFSET ?",
(source, max_limit + 1, offset)
).fetchall()
conn.close()
has_more = len(rows) > max_limit
msgs = []
for r in rows[:max_limit]:
ts = r['created_at']
if isinstance(ts, (int, float)):
time_str = datetime.fromtimestamp(ts).strftime('%m-%d %H:%M')
else:
time_str = str(ts)[:16] if ts else ''
content = r['content'] or ''
if len(content) > 3000:
content = content[:3000] + '…'
msgs.append({
'id': r['id'],
'role': r['role'],
'content': content,
'created_at': time_str,
})
msgs.reverse()
return {'messages': msgs, 'has_more': has_more}
except Exception as e:
return {'messages': [], 'has_more': False, 'error': str(e)}
def load_chat_history_by_source(source, offset=0, limit=20):
"""按 source 直接查聊天历史（兼容旧版前端，不用 task_id 关联）"""
max_limit = min(limit, 50)
try:
conn = sqlite3.connect(CHAT_HISTORY_DB)
conn.row_factory = sqlite3.Row
rows = conn.execute(
"SELECT id, role, content, created_at FROM messages "
"WHERE source=? AND role IN ('user', 'assistant') "
"ORDER BY id DESC LIMIT ? OFFSET ?",
(source, max_limit + 1, offset)
).fetchall()
conn.close()
has_more = len(rows) > max_limit
msgs = []
for r in rows[:max_limit]:
ts = r['created_at']
if isinstance(ts, (int, float)):
time_str = datetime.fromtimestamp(ts).strftime('%m-%d %H:%M')
else:
time_str = str(ts)[:16] if ts else ''
content = r['content'] or ''
if len(content) > 3000:
content = content[:3000] + '…'
msgs.append({
'id': r['id'],
'role': r['role'],
'content': content,
'created_at': time_str,
})
msgs.reverse()
return {'messages': msgs, 'has_more': has_more}
except Exception as e:
return {'messages': [], 'has_more': False, 'error': str(e)}
def get_chat_msg_count(task_id):
source = f"task-{task_id}"
try:
conn = sqlite3.connect(CHAT_HISTORY_DB)
cnt = conn.execute(
"SELECT COUNT(*) FROM messages WHERE source=?", (source,)
).fetchone()[0]
conn.close()
return cnt
except Exception:
return 0
def update_chat_message(msg_id, new_content):
"""编辑消息内容"""
try:
conn = sqlite3.connect(CHAT_HISTORY_DB)
conn.execute(
"UPDATE messages SET content=?, created_at=? WHERE id=?",
(new_content, datetime.now(timezone.utc).timestamp(), msg_id)
)
affected = conn.total_changes
conn.commit()
conn.close()
return affected > 0
except Exception:
return False
def delete_chat_message(msg_id):
"""删除消息（软删除：置空内容）"""
try:
conn = sqlite3.connect(CHAT_HISTORY_DB)
conn.execute(
"UPDATE messages SET content='[消息已删除]', created_at=? WHERE id=?",
(datetime.now(timezone.utc).timestamp(), msg_id)
)
affected = conn.total_changes
conn.commit()
conn.close()
return affected > 0
except Exception:
return False
def get_chat_message_by_id(msg_id):
"""获取单条消息"""
try:
conn = sqlite3.connect(CHAT_HISTORY_DB)
conn.row_factory = sqlite3.Row
row = conn.execute(
"SELECT id, source, role, content, created_at FROM messages WHERE id=?", (msg_id,)
).fetchone()
conn.close()
if row:
return {
'id': row['id'],
'source': row['source'],
'role': row['role'],
'content': row['content'] or '',
'created_at': row['created_at'],
}
return None
except Exception:
return None
def search_chat_messages(task_id, keyword, limit=20):
"""搜索聊天消息"""
source = f"task-{task_id}"
max_limit = min(limit, 50)
try:
conn = sqlite3.connect(CHAT_HISTORY_DB)
conn.row_factory = sqlite3.Row
rows = conn.execute(
"SELECT id, role, content, created_at FROM messages "
"WHERE source=? AND role IN ('user','assistant') AND content LIKE ? "
"ORDER BY id DESC LIMIT ?",
(source, f'%{keyword}%', max_limit)
).fetchall()
conn.close()
msgs = []
for r in rows:
ts = r['created_at']
if isinstance(ts, (int, float)):
time_str = datetime.fromtimestamp(ts).strftime('%m-%d %H:%M')
else:
time_str = str(ts)[:16] if ts else ''
content = r['content'] or ''
msgs.append({
'id': r['id'],
'role': r['role'],
'content': content,
'created_at': time_str,
})
msgs.reverse()
return {'messages': msgs, 'count': len(msgs)}
except Exception as e:
return {'messages': [], 'count': 0, 'error': str(e)}
def get_task_list():
try:
conn = sqlite3.connect(CONTENT_DB)
conn.row_factory = sqlite3.Row
rows = conn.execute(
"SELECT t.id, t.name, t.status, t.folder_id, t.created_at, t.updated_at, t.dependencies "
"FROM tasks t ORDER BY t.updated_at DESC"
).fetchall()
conn.close()
tasks = []
for r in rows:
tasks.append({
'id': r['id'],
'name': r['name'],
'status': r['status'],
'folder_id': r['folder_id'],
'created_at': r['created_at'],
'updated_at': r['updated_at'],
'dependencies': r['dependencies'] or '',
'msg_count': 0,
})
for t in tasks:
t['msg_count'] = get_chat_msg_count(t['id'])
return tasks
except Exception as e:
return []
def get_task_folder_tree(folder_id):
try:
conn = sqlite3.connect(CONTENT_DB)
conn.row_factory = sqlite3.Row
tree = _build_folder_tree(conn, folder_id)
conn.close()
return tree
except Exception:
return {'error': str(Exception)}
def get_task_folders():
"""返回文件夹分层结构，每个文件夹下挂载该文件夹的 tasks"""
try:
conn = sqlite3.connect(CONTENT_DB)
conn.row_factory = sqlite3.Row
# 获取所有文件夹
folders = conn.execute(
"SELECT id, name, parent_id FROM folders ORDER BY name"
).fetchall()
# 获取所有 tasks（active / running / created）
tasks = conn.execute(
"SELECT id, name, status, folder_id, updated_at FROM tasks "
"WHERE status IN ('active','running','created','completed') "
"ORDER BY updated_at DESC"
).fetchall()
conn.close()
# 按 folder_id 分组 tasks
tasks_by_folder = {}
for t in tasks:
fid = t['folder_id'] or ''
if fid not in tasks_by_folder:
tasks_by_folder[fid] = []
tasks_by_folder[fid].append({
'id': t['id'],
'name': t['name'],
'status': t['status'],
})
# 按 parent_id 分组文件夹
children_by_parent = {}
folder_map = {}
for f in folders:
fid = f['id']
parent = f['parent_id'] or '-1'
if parent not in children_by_parent:
children_by_parent[parent] = []
children_by_parent[parent].append(f)
folder_map[fid] = {
'id': fid,
'name': f['name'],
'parent_id': parent,
}
def build_node(fid, depth=0):
node = {
'id': fid,
'name': folder_map.get(fid, {}).get('name', str(fid)),
'type': 'folder',
'tasks': tasks_by_folder.get(fid, []),
'children': [],
}
for child in children_by_parent.get(fid, []):
node['children'].append(build_node(child['id'], depth + 1))
return node
# 根节点
roots = []
for f in folders:
parent = f['parent_id'] or '-1'
if parent == '-1':
roots.append(build_node(f['id']))
# 处理没有父文件夹的顶级 folder_id（如 '私有库/tasks/')
orphan_tasks = tasks_by_folder.get('', [])
orphan_tasks.extend(tasks_by_folder.get(None, []))
extra_keys = [k for k in tasks_by_folder if k and k not in folder_map and k != '']
return {
'roots': roots,
'orphan_tasks': orphan_tasks,
'extra_folders': extra_keys, # folder_id 值在 folders 表中找不到的
}
except Exception as e:
return {'error': str(e), 'roots': [], 'orphan_tasks': []}
def _build_folder_tree(conn, folder_id, max_depth=2, depth=0):
folders = conn.execute(
"SELECT id, name FROM folders WHERE parent_id=? ORDER BY name",
(folder_id,)
).fetchall()
docs = conn.execute(
"SELECT id, title FROM documents WHERE folder_id=? ORDER BY title",
(folder_id,)
).fetchall()
children = []
for f in folders:
if depth < max_depth - 1:
sub = _build_folder_tree(conn, f['id'], max_depth, depth + 1)
else:
sub = [] # 超过最大深度时不展开子节点
children.append({
'type': 'folder', 'id': f['id'], 'name': f['name'],
'children': sub,
})
for d in docs:
children.append({
'type': 'doc', 'id': d['id'], 'name': d['title'],
})
return children
# ═══════════════════════════════════════════
# JSON-RPC 客户端 — 连接 Hermes Dashboard
# ═══════════════════════════════════════════
class HermesRpcClient:
"""JSON-RPC WebSocket 客户端，连接 Hermes Dashboard /api/ws"""
def __init__(self):
self._ws = None
self._rpc_id = 0
self._session_id = None
self._pending = {} # rpc_id -> asyncio.Future
self._running = False
self._token = _fetch_dashboard_token()
self._sid_counter = {}
self._ws_lock = asyncio.Lock() # 防止并发 recv
async def _connect(self):
"""建立 WebSocket 连接并等待 gateway.ready"""
token = self._token
uri = f"{HERMES_DASHBOARD_WS}?token={token}"
self._ws = await websockets.connect(uri, max_size=2**20)
# 等待 gateway.ready
msg = await asyncio.wait_for(self._ws.recv(), timeout=10)
data = json.loads(msg)
assert data.get('method') == 'event' and data.get('params', {}).get('type') == 'gateway.ready', \
f"Expected gateway.ready, got: {msg[:200]}"
print(f"[RPC] Connected, gateway ready — skin: {data['params'].get('payload', {}).get('skin')}", file=sys.stderr)
return data
async def call(self, method, params=None):
"""发送 JSON-RPC 请求，等待 result 返回"""
async with self._ws_lock:
self._rpc_id += 1
rpc_id = self._rpc_id
req = {
'jsonrpc': '2.0',
'method': method,
'params': params or {},
'id': rpc_id,
}
await self._ws.send(json.dumps(req))
# 读取直到拿到 result
while True:
msg = await asyncio.wait_for(self._ws.recv(), timeout=30)
data = json.loads(msg)
if data.get('id') == rpc_id:
if 'result' in data:
return data['result']
elif 'error' in data:
raise Exception(f"RPC error: {data['error']}")
# 其他事件忽略（等待 call 完成）
async def create_session(self, source='webchat'):
"""创建新会话，返回 dict: {session_id, stored_session_id}"""
result = await self.call('session.create', {
'source': source,
})
routing_sid = result.get('session_id', '')
stored_sid = result.get('stored_session_id', routing_sid) # 兼容旧版
self._session_id = routing_sid
print(f"[RPC] Session created: routing={routing_sid} stored={stored_sid}", file=sys.stderr)
return {'session_id': routing_sid, 'stored_session_id': stored_sid}
async def _read_until_timeout(self, timeout=5.0):
"""读取一条消息，最多等 timeout 秒。返回 dict 或 None（超时）"""
try:
msg = await asyncio.wait_for(self._ws.recv(), timeout=timeout)
return json.loads(msg)
except asyncio.TimeoutError:
return None
except websockets.exceptions.ConnectionClosed:
return None
async def submit_prompt_stream(self, sid, text, event_callback):
"""提交 prompt 并流式读取事件，event_callback(event_type, payload)
整个流式过程持有 _ws_lock，防止并发 recv 冲突。
不同 task 的 consumer 会排队等待锁，不会报错。
"""
async with self._ws_lock:
# 直接发送 prompt.submit（不使用 call()，因为我们需持有锁进入流式读取）
self._rpc_id += 1
rpc_id = self._rpc_id
req = {
'jsonrpc': '2.0',
'method': 'prompt.submit',
'params': {'session_id': sid, 'text': text},
'id': rpc_id,
}
await self._ws.send(json.dumps(req))
# 读取直到拿到 prompt.submit 的 response
result = None
while True:
msg = await asyncio.wait_for(self._ws.recv(), timeout=30)
data = json.loads(msg)
if data.get('id') == rpc_id:
if 'result' in data:
result = data['result']
break
elif 'error' in data:
raise Exception(f"RPC error: {data['error']}")
# 校验结果
if not isinstance(result, dict) or result.get('status') != 'streaming':
raise Exception(f"prompt.submit failed: {result}")
# 持有锁 → 读取事件流直到 message.complete
tool_calls = []
while True:
data = await self._read_until_timeout(timeout=120)
if data is None:
break
# 跳过 RPC response
if data.get('id') is not None:
continue
params = data.get('params', {})
if not isinstance(params, dict):
continue
evt_type = params.get('type', '')
payload = params.get('payload', {})
if not evt_type:
continue
# 调用回调
if callable(event_callback):
await event_callback(evt_type, payload)
if evt_type == 'message.complete':
return payload
if evt_type == 'tool.start':
tool_calls.append(payload.get('name', ''))
return {'text': 'timeout'}
# ═══════════════════════════════════════════
# WebSocket 处理（对外）
# ═══════════════════════════════════════════
async def handle_ws(websocket):
conn_info = {'device_id': '', 'current_task_id': None}
active_connections[websocket] = conn_info
try:
async for raw in websocket:
try:
data = json.loads(raw)
except json.JSONDecodeError:
continue
msg_type = data.get('type', '')
if msg_type == 'auth':
conn_info['device_id'] = data.get('device_id', '')
await ws_send(websocket, {'type': 'ack'})
elif msg_type == 'select_task':
task_id = data.get('task_id', '')
conn_info['current_task_id'] = task_id
await ws_send(websocket, {'type': 'task_selected', 'task_id': task_id})
elif msg_type == 'message':
msg_id = data.get('msg_id', 0)
content = data.get('content', '').strip()
task_id = data.get('task_id', '') or conn_info.get('current_task_id', '')
if not content or not task_id:
continue
asyncio.create_task(
enqueue_chat_message(websocket, msg_id, content, task_id)
)
except asyncio.CancelledError:
pass
except Exception:
pass
finally:
active_connections.pop(websocket, None)
# ═══════════════════════════════════════════
# 消息队列调度器
# ═══════════════════════════════════════════
async def enqueue_chat_message(ws, msg_id, content, task_id):
global chat_queue_msg_id_counter
chat_queue_msg_id_counter += 1
queue_id = chat_queue_msg_id_counter
# ── 微服务路由预检：记录匹配的微服务到 task_board ──
matched_services = []
try:
conn = sqlite3.connect(CONTENT_DB)
rows = conn.execute("SELECT service_name, domain_keywords FROM service_agents WHERE status='active'").fetchall()
conn.close()
for name, kw_json in rows:
keywords = json.loads(kw_json) if kw_json else []
if any(kw.lower() in content.lower() for kw in keywords):
matched_services.append(name)
except Exception:
pass # 预检失败不阻塞消息
if task_id not in chat_queues:
chat_queues[task_id] = asyncio.Queue()
q = chat_queues[task_id]
item = {
'ws': ws,
'msg_id': msg_id,
'content': content,
'task_id': task_id,
'queue_id': queue_id,
}
await q.put(item)
# 消息已通过 RPC submit 发给内核，内核自动存到 state.db
# 不再写 chat_history.db（冗余）
task_key = f"task-{queue_id}"
svc_tag = f" 🏛️{','.join(matched_services)}" if matched_services else ""
task_board[task_key] = {
'status': f'排队中{svc_tag}',
'content': content[:60],
'created_at': _time_module.time(),
'elapsed': 0,
'queue_pos': q.qsize(),
}
task_board_order.append(task_key)
_trim_task_board()
if not chat_queue_busy.get(task_id):
chat_queue_busy[task_id] = True
asyncio.create_task(chat_queue_consumer(task_id))
def _trim_task_board():
while len(task_board_order) > 20:
old_id = task_board_order.pop(0)
task_board.pop(old_id, None)
async def chat_queue_consumer(task_id):
while True:
q = chat_queues.get(task_id)
if not q:
chat_queue_busy[task_id] = False
return
try:
item = await asyncio.wait_for(q.get(), timeout=3600)
except asyncio.TimeoutError:
chat_queue_busy[task_id] = False
return
ws = item['ws']
msg_id = item['msg_id']
content = item['content']
tid = item['task_id']
queue_id = item['queue_id']
task_key = f"task-{queue_id}"
if task_key in task_board:
task_board[task_key]['status'] = '执行中'
task_board[task_key]['created_at'] = _time_module.time()
remaining = q.qsize()
for tk in task_board_order:
if task_board[tk]['status'] == '排队中':
task_board[tk]['queue_pos'] = remaining
if remaining > 0:
await send_local_reply(ws, msg_id,
f'📋 收到！当前还有 {remaining} 条消息在排队，按顺序处理中…',
task_id=tid)
await run_hermes_rpc(ws, msg_id, content, tid)
if task_key in task_board:
task_board[task_key]['status'] = '已完成'
task_board[task_key]['created_at'] = _time_module.time()
async def send_local_reply(ws, msg_id, text, task_id=''):
await ws_send(ws, {
'type': 'reply_done',
'msg_id': msg_id,
'task_id': task_id,
'reply': text,
})
async def ws_send(ws, data):
try:
await ws.send(json.dumps(data, ensure_ascii=False))
except Exception:
pass
# ═══════════════════════════════════════════
# JSON-RPC 方式调用 Hermes
# ═══════════════════════════════════════════
_rpc_client_holder = None
_rpc_client_lock = asyncio.Lock()
async def _get_hermes_rpc():
"""获取 RPC 客户端（单例，自动重连）"""
global _rpc_client_holder
async with _rpc_client_lock:
if _rpc_client_holder is None:
_rpc_client_holder = HermesRpcClient()
await _rpc_client_holder._connect()
# 如果连接断了，重新创建
if _rpc_client_holder._ws is None or _rpc_client_holder._ws.close_code is not None:
_rpc_client_holder = HermesRpcClient()
await _rpc_client_holder._connect()
return _rpc_client_holder
async def run_hermes_rpc(ws, msg_id, prompt, task_id):
"""使用 JSON-RPC 调用 Hermes，流式推送事件给前端"""
try:
# 获取任务上下文
task_name = task_id
task_context = ""
try:
conn = sqlite3.connect(CONTENT_DB)
row = conn.execute("SELECT name, description FROM tasks WHERE id=?", (task_id,)).fetchone()
if row:
task_name = row[0]
if row[1]:
task_context = row[1][:200]
conn.close()
except Exception:
pass
# 构建增强 prompt
enhanced_prompt = f"[当前任务] {task_name}\n[任务ID] {task_id}\n\n{prompt}"
# 获取 RPC 客户端
rpc = await _get_hermes_rpc()
# 复用 task 对应的内核 session（持久化到 content.db）
stored_sid = _ensure_valid_session_id(task_id) or get_task_session_id(task_id)
if stored_sid:
routing_sid = get_task_routing_session_id(task_id) or stored_sid
print(f"[RPC] Reuse session stored={stored_sid} routing={routing_sid}", file=sys.stderr)
else:
session_info = await rpc.create_session(source=f'task-{task_id}')
routing_sid = session_info['session_id']
stored_sid = session_info['stored_session_id']
save_task_session_id(task_id, stored_sid, routing_sid)
print(f"[RPC] New session stored={stored_sid} routing={routing_sid} (persisted)", file=sys.stderr)
# 迁移 chat_history.db 中的旧消息到新 session
try:
ch_db = CHAT_HISTORY_DB
if os.path.exists(ch_db):
ch_conn = sqlite3.connect(ch_db)
ch_rows = ch_conn.execute(
"SELECT role, content, created_at FROM messages WHERE source=? ORDER BY id",
(f"task-{task_id}",)
).fetchall()
ch_conn.close()
if ch_rows:
st_conn = sqlite3.connect(os.path.expanduser('~/.hermes/state.db'))
st_conn.execute("BEGIN")
for row in ch_rows:
st_conn.execute(
"INSERT INTO messages (session_id, role, content, timestamp, active) "
"SELECT ?, ?, ?, ?, 1 "
"WHERE NOT EXISTS ("
" SELECT 1 FROM messages WHERE session_id=? AND role=? AND content=? AND timestamp=?"
")",
(stored_sid, row[0], row[1] or '', float(row[2]) if isinstance(row[2], (int, float)) else 0,
stored_sid, row[0], row[1] or '', float(row[2]) if isinstance(row[2], (int, float)) else 0)
)
st_conn.execute(
"UPDATE sessions SET message_count = (SELECT COUNT(*) FROM messages WHERE session_id=?) WHERE id=?",
(stored_sid, stored_sid)
)
st_conn.commit()
st_conn.close()
print(f"[MIGRATE] Migrated {len(ch_rows)} old messages from chat_history.db to session {stored_sid}", file=sys.stderr)
except Exception as e:
print(f"[MIGRATE] Error: {e}", file=sys.stderr)
# 收集完整回复用于保存
full_text = ""
full_reasoning = ""
tool_calls_info = []
async def on_event(evt_type, payload):
nonlocal full_text, full_reasoning
# 所有 WS 消息都带上 task_id，方便前端按任务隔离
def msg(**kw):
kw.setdefault('task_id', task_id)
return kw
if evt_type == 'message.start':
# 发送推理面板开始信号
await ws_send(ws, msg(**{
'type': 'reply_chunk',
'msg_id': msg_id,
'chunk': '',
'stage': 'thinking_start',
}))
elif evt_type == 'reasoning.delta':
text = payload.get('text', '')
if text:
full_reasoning += text
await ws_send(ws, msg(**{
'type': 'reasoning_delta',
'msg_id': msg_id,
'text': text,
}))
elif evt_type == 'thinking.delta':
text = payload.get('text', '') or ''
if text.strip():
await ws_send(ws, msg(**{
'type': 'reply_chunk',
'msg_id': msg_id,
'chunk': text + '\n',
'stage': 'thinking',
}))
elif evt_type == 'message.delta':
text = payload.get('text', '')
if text:
full_text += text
await ws_send(ws, msg(**{
'type': 'reply_chunk',
'msg_id': msg_id,
'chunk': text,
'stage': 'streaming',
}))
elif evt_type == 'tool.start':
name = payload.get('name', 'tool')
args = payload.get('args', {})
tool_info = {
'name': name,
'args': args,
'status': 'running',
}
tool_calls_info.append(tool_info)
await ws_send(ws, msg(**{
'type': 'reply_chunk',
'msg_id': msg_id,
'chunk': f'\n\n🔧 {name}\n',
'stage': 'tool_start',
'tool': {
'name': name,
'args': args,
},
}))
elif evt_type == 'tool.complete':
if tool_calls_info:
tool_calls_info[-1]['status'] = 'done'
await ws_send(ws, msg(**{
'type': 'reply_chunk',
'msg_id': msg_id,
'chunk': '',
'stage': 'tool_complete',
}))
elif evt_type == 'reasoning.available':
text = payload.get('text', '')
if text and not full_reasoning:
full_reasoning = text
await ws_send(ws, msg(**{
'type': 'reasoning_delta',
'msg_id': msg_id,
'text': text,
}))
await ws_send(ws, msg(**{
'type': 'reasoning_done',
'msg_id': msg_id,
}))
elif evt_type.startswith('subagent.'):
# 子代理事件 — 转发给前端
name = payload.get('goal', '子任务')
task_index = payload.get('task_index', 0)
task_count = payload.get('task_count', 1)
subagent_id = payload.get('subagent_id', '')
text = payload.get('text', '')
tool_name = payload.get('tool_name', '')
status = payload.get('status', '')
summary = payload.get('summary', '')
depth = payload.get('depth', 0)
output_tokens = payload.get('output_tokens', 0)
input_tokens = payload.get('input_tokens', 0)
if evt_type == 'subagent.start':
await ws_send(ws, msg(**{
'type': 'subagent_event',
'msg_id': msg_id,
'subagent_type': 'start',
'name': name,
'task_index': task_index,
'task_count': task_count,
'subagent_id': subagent_id,
'text': text,
}))
elif evt_type == 'subagent.thinking':
if text:
await ws_send(ws, msg(**{
'type': 'subagent_event',
'msg_id': msg_id,
'subagent_type': 'thinking',
'text': text,
'subagent_id': subagent_id,
}))
elif evt_type == 'subagent.text':
if text:
await ws_send(ws, msg(**{
'type': 'subagent_event',
'msg_id': msg_id,
'subagent_type': 'text',
'text': text,
'subagent_id': subagent_id,
}))
elif evt_type == 'subagent.tool':
await ws_send(ws, msg(**{
'type': 'subagent_event',
'msg_id': msg_id,
'subagent_type': 'tool',
'tool_name': tool_name or name,
'text': text,
'subagent_id': subagent_id,
}))
elif evt_type == 'subagent.complete':
await ws_send(ws, msg(**{
'type': 'subagent_event',
'msg_id': msg_id,
'subagent_type': 'complete',
'name': name,
'summary': summary,
'subagent_id': subagent_id,
'input_tokens': input_tokens,
'output_tokens': output_tokens,
}))
# 提交 prompt 并流式获取事件（带 session 失效重试）
max_session_retry = 1
for attempt in range(max_session_retry + 1):
try:
complete_payload = await rpc.submit_prompt_stream(routing_sid, enhanced_prompt, on_event)
break # 成功，跳出重试
except Exception as e:
error_str = str(e)
is_session_404 = 'session not found' in error_str or 'code: 4001' in error_str or '4001' in error_str
if is_session_404 and attempt < max_session_retry:
print(f"[RPC] Session 404 for task {task_id}, recreating and retrying...", file=sys.stderr)
# 清空失效的 session
try:
conn = sqlite3.connect(CONTENT_DB)
conn.execute("UPDATE tasks SET session_id=NULL, routing_session_id=NULL WHERE id=?", (task_id,))
conn.commit()
conn.close()
except Exception:
pass
# 重新创建 session
session_info = await rpc.create_session(source=f'task-{task_id}')
routing_sid = session_info['session_id']
stored_sid = session_info['stored_session_id']
save_task_session_id(task_id, stored_sid, routing_sid)
# 重置闭包变量，on_event 使用的将是新值
full_text = ""
full_reasoning = ""
tool_calls_info = []
print(f"[RPC] Retry with new session routing={routing_sid}", file=sys.stderr)
continue
raise # 其他错误或已达重试上限，交给外层 except
# 如果是 message.complete 返回的，提取最终文本
# 注意：message.complete 的 payload.text 可能只包含最后一段 delta，
# 所以优先使用 full_text（通过 message.delta 累积的完整回复）
if complete_payload and isinstance(complete_payload, dict):
final_text = full_text or complete_payload.get('text', '')
status = complete_payload.get('status', 'complete')
else:
final_text = full_text
status = 'complete'
# 发送推理结束信号
if full_reasoning:
await ws_send(ws, {
'type': 'reasoning_done',
'msg_id': msg_id,
})
# 发送完成信号
await ws_send(ws, {
'type': 'reply_done',
'msg_id': msg_id,
'task_id': task_id,
'reply': final_text,
})
# 保存 AI 回复（内核已自动存到 state.db，这里不再重复写 chat_history.db）
# 保留注释，如果后续需要做 task 关联分析可恢复
if final_text.strip():
pass
except Exception as e:
import traceback
traceback.print_exc()
await ws_send(ws, {
'type': 'error',
'msg_id': msg_id,
'task_id': task_id,
'message': str(e),
})
# ═══════════════════════════════════════════
# HTTP API 处理
# ═══════════════════════════════════════════
def handle_api(method, path, headers, body):
parsed = urllib.parse.urlparse(path)
params = urllib.parse.parse_qs(parsed.query)
if path == '/doc-name' or path.startswith('/doc-name?'):
doc_id = params.get('doc_id', [''])[0]
if not doc_id:
return (200, {'name': ''})
try:
conn = sqlite3.connect(CONTENT_DB)
row = conn.execute("SELECT title FROM documents WHERE id=?", (doc_id,)).fetchone()
conn.close()
return (200, {'name': row[0] if row else ''})
except Exception as e:
return (200, {'name': ''})
if path == '/chat/tasks' or path.startswith('/chat/tasks?'):
tasks = get_task_list()
return (200, {'tasks': tasks})
elif path == '/chat/task-folders' or path.startswith('/chat/task-folders?'):
result = get_task_folders()
return (200, result)
elif path == '/chat/task/touch' and method == 'POST':
try:
body_data = json.loads(body) if body else {}
task_id = body_data.get('task_id', '')
if task_id:
touch_task_api(task_id)
return (200, {'ok': True, 'task_id': task_id})
return (400, {'error': 'task_id required'})
except Exception as e:
return (400, {'error': str(e)})
elif path == '/chat/wechat/hook' and method == 'POST':
"""
微信消息后处理 hook。
微信消息回复完成后，由 Hermes Gateway 回调此接口（或我主动调用）。
body: {
"content": "用户说的内容",
"reply": "我的回复",
"source": "weixin"
}
功能: 意图识别 → 关联 task → 保存消息到 chat_history.db → 更新 updated_at
"""
try:
body_data = json.loads(body) if body else {}
content = body_data.get('content', '')
reply = body_data.get('reply', '')
# 从 content.db 加载所有 active 的 task
conn = sqlite3.connect(CONTENT_DB)
conn.row_factory = sqlite3.Row
rows = conn.execute(
"SELECT id, name, keywords, description FROM tasks WHERE status IN ('active','running','completed','created')"
).fetchall()
conn.close()
matched_task = None
best_score = 0
# 简化的意图匹配：keywords 和 description 关键词匹配
for row in rows:
task_id = row['id']
keywords = (row['keywords'] or '').lower()
desc = (row['description'] or '').lower()
name = (row['name'] or '').lower()
combined = f"{keywords} {desc} {name}"
content_lower = content.lower()
# 计算匹配分数
score = 0
# 关键词精确匹配
for kw in combined.split():
kw = kw.strip()
if kw and len(kw) > 1 and kw in content_lower:
score += 3
# 全名/ID精确匹配
if name and name in content_lower:
score += 10
if task_id in content_lower:
score += 10
if score > best_score:
best_score = score
matched_task = task_id
result = {'matched': False, 'task_id': None}
# 如果匹配到 task（阈值>=3）
if matched_task and best_score >= 3:
# 保存消息到 chat_history.db
now = datetime.now(timezone.utc).timestamp()
source = f"task-{matched_task}"
try:
conn = sqlite3.connect(CHAT_HISTORY_DB)
conn.execute(
"INSERT INTO messages (source, role, content, created_at) VALUES (?, ?, ?, ?)",
(source, 'user', content, now)
)
if reply:
conn.execute(
"INSERT INTO messages (source, role, content, created_at) VALUES (?, ?, ?, ?)",
(source, 'assistant', reply, now + 0.001)
)
conn.commit()
conn.close()
except Exception:
pass
# 更新 task 时间
touch_task_api(matched_task)
result = {'matched': True, 'task_id': matched_task, 'score': best_score}
return (200, result)
except Exception as e:
return (200, {'error': str(e)})
elif path == '/chat/messages/move' and method == 'POST':
"""
POST /chat/messages/move
Body: { message_id: number, target_task_id: string }
将消息移动到另一个 task 的会话
"""
try:
data = json.loads(body.decode('utf-8'))
message_id = data.get('message_id')
target_task_id = data.get('target_task_id')
if not message_id or not target_task_id:
return (400, {'error': 'message_id and target_task_id required'})
state_db_path = os.path.expanduser('~/.hermes/state.db')
conn = sqlite3.connect(state_db_path)
# 1. 获取消息当前的 session_id
row = conn.execute("SELECT session_id FROM messages WHERE id=?", (message_id,)).fetchone()
if not row:
conn.close()
return (404, {'error': 'message not found'})
source_session_id = row[0]
# 2. 获取目标 task 的 session_id
import time
conn2 = sqlite3.connect(CONTENT_DB)
target = conn2.execute("SELECT session_id FROM tasks WHERE id=?", (target_task_id,)).fetchone()
if not target:
conn2.close()
conn.close()
return (404, {'error': 'target task not found'})
target_session_id = target[0]
# 3. 如果目标 task 无 session，在 state.db 创建一条
if not target_session_id:
ts = time.time()
target_session_id = f"mv_{target_task_id[:16]}_{int(ts)}"
conn.execute(
"INSERT INTO sessions (id, source, started_at) VALUES (?, ?, ?)",
(target_session_id, f"task-{target_task_id}", ts)
)
# 持久化到 content.db
conn2.execute("UPDATE tasks SET session_id=? WHERE id=?", (target_session_id, target_task_id))
conn2.commit()
print(f"[MOVE] created session {target_session_id} for task {target_task_id}", file=sys.stderr)
conn2.close()
if source_session_id == target_session_id:
conn.close()
return (200, {'ok': True, 'moved': False, 'reason': 'already in target session'})
# 4. 移动消息
conn.execute("UPDATE messages SET session_id=? WHERE id=?", (target_session_id, message_id))
conn.commit()
conn.close()
print(f"[MOVE] message {message_id}: {source_session_id[:20]} → {target_session_id[:20]}", file=sys.stderr)
return (200, {'ok': True, 'moved': True, 'target_session_id': target_session_id})
except Exception as e:
print(f"[MOVE] error: {e}", file=sys.stderr)
return (500, {'error': str(e)})
elif path == '/documents' and method == 'GET':
try:
conn = sqlite3.connect(CONTENT_DB)
rows = conn.execute("SELECT id, title, folder_id FROM documents ORDER BY title").fetchall()
conn.close()
return (200, [{"id": r[0], "title": r[1], "folder_id": r[2]} for r in rows])
except Exception as e:
return (500, {"error": str(e)})
elif path == '/documents-with-tasks' and method == 'GET':
try:
conn = sqlite3.connect(CONTENT_DB)
docs = conn.execute("SELECT id, title, folder_id FROM documents ORDER BY title").fetchall()
rows = conn.execute("""
SELECT dt.doc_id, dt.task_id, t.name
FROM doc_tasks dt JOIN tasks t ON dt.task_id = t.id
ORDER BY dt.doc_id
""").fetchall()
conn.close()
task_map = {}
for r in rows:
doc_id = r[0]
if doc_id not in task_map:
task_map[doc_id] = []
task_map[doc_id].append({"id": r[1], "name": r[2]})
result = [{
"id": d[0],
"title": d[1],
"folder_id": d[2],
"tasks": task_map.get(d[0], [])
} for d in docs]
return (200, result)
except Exception as e:
return (500, {"error": str(e)})
elif path == '/tasks' and method == 'GET':
try:
conn = sqlite3.connect(CONTENT_DB)
rows = conn.execute("SELECT id, name FROM tasks ORDER BY name").fetchall()
conn.close()
return (200, [{"id": r[0], "name": r[1]} for r in rows])
except Exception as e:
return (500, {"error": str(e)})
elif (path == '/doc-tasks' or path.startswith('/doc-tasks?')) and method == 'GET':
doc_id = params.get('doc_id', [None])[0]
try:
conn = sqlite3.connect(CONTENT_DB)
if doc_id:
rows = conn.execute("""
SELECT t.id, t.name, t.status
FROM doc_tasks dt JOIN tasks t ON dt.task_id = t.id
WHERE dt.doc_id = ?
ORDER BY t.name
""", (doc_id,)).fetchall()
all_tasks = conn.execute("SELECT id, name FROM tasks ORDER BY name").fetchall()
conn.close()
return (200, {
"associated": [{"id": r[0], "name": r[1], "status": r[2]} for r in rows],
"all_tasks": [{"id": r[0], "name": r[1]} for r in all_tasks]
})
else:
rows = conn.execute("""
SELECT dt.doc_id, d.title, dt.task_id, t.name
FROM doc_tasks dt
JOIN documents d ON dt.doc_id = d.id
JOIN tasks t ON dt.task_id = t.id
ORDER BY d.title
""").fetchall()
conn.close()
return (200, [{"doc_id": r[0], "doc_title": r[1], "task_id": r[2], "task_name": r[3]} for r in rows])
except Exception as e:
return (500, {"error": str(e)})
elif path == '/doc-tasks' and method == 'POST':
try:
data = json.loads(body.decode('utf-8'))
doc_id = data.get('doc_id')
task_ids = data.get('task_ids', [])
conn = sqlite3.connect(CONTENT_DB)
conn.execute("DELETE FROM doc_tasks WHERE doc_id=?", (doc_id,))
for tid in task_ids:
conn.execute("INSERT OR IGNORE INTO doc_tasks (doc_id, task_id) VALUES (?, ?)", (doc_id, tid))
conn.commit()
conn.close()
return (200, {"ok": True})
except Exception as e:
return (500, {"error": str(e)})
elif path == '/tasks/create' and method == 'POST':
try:
data = json.loads(body.decode('utf-8'))
name = data.get('name', '').strip()
dir_id = data.get('dir_id') or None
if not name:
return (400, {"error": "name required"})
import uuid
task_id = 'task_' + uuid.uuid4().hex[:12]
conn = sqlite3.connect(CONTENT_DB)
conn.execute(
"INSERT INTO tasks (id, name, status, folder_id, dir_id, created_at, updated_at) VALUES (?, ?, 'created', '', ?, datetime('now','localtime'), datetime('now','localtime'))",
(task_id, name, dir_id)
)
conn.commit()
conn.close()
return (200, {"ok": True, "task_id": task_id, "name": name, "dir_id": dir_id})
except Exception as e:
return (500, {"error": str(e)})
elif path == '/tasks/rename' and method == 'POST':
try:
data = json.loads(body.decode('utf-8'))
task_id = data.get('task_id')
new_name = data.get('name')
conn = sqlite3.connect(CONTENT_DB)
conn.execute("UPDATE tasks SET name=? WHERE id=?", (new_name, task_id))
conn.commit()
conn.close()
return (200, {"ok": True})
except Exception as e:
return (500, {"error": str(e)})
elif path == '/tasks/delete' and method == 'POST':
try:
data = json.loads(body.decode('utf-8'))
task_id = data.get('task_id')
if not task_id:
return (400, {"error": "task_id required"})
conn = sqlite3.connect(CONTENT_DB)
# 清除 doc_tasks 关联
conn.execute("DELETE FROM doc_tasks WHERE task_id=?", (task_id,))
# 清除 task_dirs 挂靠
conn.execute("UPDATE tasks SET dir_id=NULL, session_id=NULL, routing_session_id=NULL WHERE id=?", (task_id,))
# 删除 task
conn.execute("DELETE FROM tasks WHERE id=?", (task_id,))
conn.commit()
conn.close()
return (200, {"ok": True})
except Exception as e:
return (500, {"error": str(e)})
elif path == '/service-agents' and method == 'GET':
try:
conn = sqlite3.connect(CONTENT_DB)
rows = conn.execute("""
SELECT sa.service_name, sa.task_id, sa.description, sa.domain_keywords, sa.status, t.name
FROM service_agents sa JOIN tasks t ON sa.task_id = t.id
ORDER BY sa.service_name
""").fetchall()
conn.close()
return (200, [{
"service_name": r[0], "task_id": r[1],
"description": r[2], "domain_keywords": json.loads(r[3]) if r[3] else [],
"status": r[4], "task_name": r[5]
} for r in rows])
except Exception as e:
return (500, {"error": str(e)})
elif path == '/service-router' and method == 'POST':
try:
data = json.loads(body.decode('utf-8'))
query = data.get('query', '')
conn = sqlite3.connect(CONTENT_DB)
rows = conn.execute("SELECT service_name, task_id, domain_keywords FROM service_agents WHERE status='active'").fetchall()
conn.close()
results = []
for name, tid, kw_json in rows:
keywords = json.loads(kw_json) if kw_json else []
match_count = sum(1 for kw in keywords if kw.lower() in query.lower())
if match_count > 0:
results.append({"service_name": name, "task_id": tid, "match_score": match_count})
results.sort(key=lambda x: x['match_score'], reverse=True)
return (200, {"matches": results, "status": "ok"})
except Exception as e:
return (500, {"error": str(e)})
elif path == '/tasks-with-folder' and method == 'GET':
try:
conn = sqlite3.connect(CONTENT_DB)
rows = conn.execute("""
SELECT t.id, t.name, t.status, t.folder_id, f.name as folder_name,
f.parent_id as folder_parent, t.dir_id
FROM tasks t
LEFT JOIN folders f ON t.folder_id = f.id
ORDER BY COALESCE(f.name, 'zzz'), t.name
""").fetchall()
conn.close()
tasks = []
for r in rows:
tasks.append({
"id": r[0], "name": r[1], "status": r[2],
"folder_id": r[3], "folder_name": r[4], "folder_parent": r[5],
"dir_id": r[6]
})
return (200, {"tasks": tasks})
except Exception as e:
return (500, {"error": str(e)})
elif path == '/task-dirs' and method == 'GET':
try:
conn = sqlite3.connect(CONTENT_DB)
rows = conn.execute("""
SELECT d.id, d.name, d.parent_id, d.sort_order,
(SELECT COUNT(*) FROM tasks t WHERE t.dir_id = d.id) as task_count
FROM task_dirs d
ORDER BY d.sort_order, d.name
""").fetchall()
conn.close()
dirs = []
for r in rows:
dirs.append({
"id": r[0], "name": r[1], "parent_id": r[2],
"sort_order": r[3], "task_count": r[4]
})
return (200, {"dirs": dirs})
except Exception as e:
return (500, {"error": str(e)})
elif path == '/task-dirs' and method == 'POST':
try:
data = json.loads(body.decode('utf-8'))
action = data.get('action', 'create')
conn = sqlite3.connect(CONTENT_DB)
if action == 'create':
import uuid
new_id = 'td_' + uuid.uuid4().hex[:12]
conn.execute("INSERT INTO task_dirs (id, name, parent_id, sort_order) VALUES (?, ?, ?, ?)",
(new_id, data['name'], data.get('parent_id'), data.get('sort_order', 0)))
conn.commit()
conn.close()
return (200, {"ok": True, "id": new_id})
elif action == 'rename':
conn.execute("UPDATE task_dirs SET name=? WHERE id=?", (data['name'], data['id']))
conn.commit()
conn.close()
return (200, {"ok": True})
elif action == 'move_task':
conn.execute("UPDATE tasks SET dir_id=? WHERE id=?", (data.get('dir_id'), data['task_id']))
conn.commit()
conn.close()
return (200, {"ok": True})
elif action == 'move_dir':
dir_id = data.get('id')
new_parent = data.get('parent_id')
if not dir_id or new_parent is None:
conn.close()
return (400, {"error": "id and parent_id required"})
if dir_id == new_parent:
conn.close()
return (400, {"error": "不能将目录移到自己下面"})
# 防循环检查
def is_descendant(pid):
for row in conn.execute("SELECT id FROM task_dirs WHERE parent_id=?", (pid,)):
if row[0] == dir_id or is_descendant(row[0]):
return True
return False
if is_descendant(new_parent):
conn.close()
return (400, {"error": "不能将目录移到自己的子目录下（会导致循环）"})
conn.execute("UPDATE task_dirs SET parent_id=? WHERE id=?", (new_parent, dir_id))
conn.commit()
conn.close()
return (200, {"ok": True})
elif action == 'delete':
# 不能删除有子目录的节点
child = conn.execute("SELECT COUNT(*) FROM task_dirs WHERE parent_id=?", (data['id'],)).fetchone()[0]
if child > 0:
conn.close()
return (400, {"error": "请先删除子目录"})
# 解绑tasks
conn.execute("UPDATE tasks SET dir_id=NULL WHERE dir_id=?", (data['id'],))
conn.execute("DELETE FROM task_dirs WHERE id=?", (data['id'],))
conn.commit()
conn.close()
return (200, {"ok": True})
conn.close()
return (400, {"error": "unknown action"})
except Exception as e:
return (500, {"error": str(e)})
elif path.startswith('/chat/tree'):
folder_id = params.get('folder_id', [''])[0]
task_id = params.get('task_id', [''])[0]
if task_id and not folder_id:
# 通过 task_id 查：只返回 doc_tasks 关联的文档（按文件夹组织）
try:
conn = sqlite3.connect(CONTENT_DB)
conn.row_factory = sqlite3.Row
rows = conn.execute(
"""SELECT d.id as doc_id, d.title as doc_title, d.folder_id,
f.name as folder_name
FROM doc_tasks dt
JOIN documents d ON d.id = dt.doc_id
LEFT JOIN folders f ON f.id = d.folder_id
WHERE dt.task_id=?
ORDER BY d.folder_id, d.title""",
(task_id,)
).fetchall()
conn.close()
# 按 folder 分组
folder_docs = {}
for r in rows:
fid = r['folder_id']
if fid not in folder_docs:
folder_docs[fid] = {
'type': 'folder', 'id': fid, 'name': r['folder_name'] or str(fid),
'children': [],
}
folder_docs[fid]['children'].append({
'type': 'doc', 'id': r['doc_id'], 'name': r['doc_title'],
})
tree = list(folder_docs.values())
return (200, {'tree': tree})
except Exception as e:
print(f"[TREE] error for task_id={task_id}: {e}", file=sys.stderr)
return (200, {'tree': []})
# folder_id 模式：传统文件夹树浏览
if not folder_id and task_id:
try:
conn = sqlite3.connect(CONTENT_DB)
row = conn.execute(
"SELECT folder_id FROM tasks WHERE id=?", (task_id,)
).fetchone()
conn.close()
if row:
folder_id = row[0]
except Exception:
pass
if not folder_id:
return (200, {'tree': []})
tree = get_task_folder_tree(folder_id)
return (200, {'tree': tree})
elif path.startswith('/chat/history'):
task_id = params.get('task_id', [''])[0]
# 兼容旧版前端用 source 参数查历史
source = params.get('source', [''])[0]
offset = int(params.get('offset', ['0'])[0])
limit = min(int(params.get('limit', ['20'])[0]), 50)
if task_id:
# 新版：通过 kernel state.db 查（session_id 持久化在 tasks 表）
result = load_chat_history_from_kernel(task_id, offset, limit)
return (200, result)
if source == 'chat' and not task_id:
# source=chat（WebChat 首页）：自动解析到最活跃 task
auto_task_id = resolve_task_for_source('chat')
if auto_task_id:
print(f"[HISTORY] auto-resolved source=chat -> task {auto_task_id}", file=sys.stderr)
result = load_chat_history_from_kernel(auto_task_id, offset, limit)
return (200, result)
if source and not task_id:
# 旧版兼容：source 模式查 chat_history.db
result = load_chat_history_by_source(source, offset, limit)
return (200, result)
return (200, {'messages': [], 'has_more': False})
elif path.startswith('/chat/current-task'):
# 返回当前活跃 task（WebChat 默认关联的任务）
task_id = resolve_task_for_source('chat')
if task_id:
try:
conn = sqlite3.connect(CONTENT_DB)
row = conn.execute(
"SELECT id, name, status, session_id FROM tasks WHERE id=?",
(task_id,)
).fetchone()
conn.close()
if row:
return (200, {
'task_id': row[0],
'name': row[1],
'status': row[2],
'session_id': row[3] or '',
})
except Exception:
pass
return (200, {'task_id': None})
elif path.startswith('/chat/prd'):
prd_path = os.path.join(os.path.dirname(__file__), 'PRD.md')
if os.path.exists(prd_path):
with open(prd_path, 'r', encoding='utf-8') as f:
content = f.read()
return (200, content, 'text/markdown')
return (200, '# PRD\n\nPRD doc not found', 'text/markdown')
elif path == '/chat/message/retain' and method == 'POST':
"""前端发消息时先保留到 state.db，防止切 task 丢消息"""
try:
data = json.loads(body.decode('utf-8'))
task_id = data.get('task_id', '')
role = data.get('role', 'user')
content = data.get('content', '').strip()
if not task_id or not content:
return (400, {'error': 'task_id and content required'})
import time
state_db = os.path.expanduser('~/.hermes/state.db')
conn = sqlite3.connect(state_db)
# 获取已有的 session_id
session_id = get_task_session_id(task_id)
if not session_id:
# 创建合成 session（防止外键约束）
ts = time.time()
session_id = f"retain_{task_id[:16]}_{int(ts)}"
conn.execute(
"INSERT OR IGNORE INTO sessions (id, source, started_at) VALUES (?, ?, ?)",
(session_id, f"task-{task_id}", ts)
)
# 持久化到 content.db
save_task_session_id(task_id, session_id, session_id)
# 写入消息
conn.execute(
"INSERT INTO messages (session_id, role, content, timestamp, active) VALUES (?, ?, ?, ?, 1)",
(session_id, role, content, time.time())
)
# 更新计数
conn.execute(
"UPDATE sessions SET message_count = (SELECT COUNT(*) FROM messages WHERE session_id=?) WHERE id=?",
(session_id, session_id)
)
conn.commit()
conn.close()
return (200, {'ok': True, 'session_id': session_id})
except Exception as e:
print(f"[RETAIN ERROR] {e}", file=sys.stderr)
return (500, {'error': str(e)})
elif path.startswith('/chat/attachment/') and method == 'POST' and path.endswith('/upload'):
"""附件上传"""
import time, subprocess, re, secrets
# 生成 OID: 24字符hex (8位时间戳 + 16位随机)
ts_hex = format(int(time.time()), '08x')
rand_hex = secrets.token_hex(8) # 16 hex chars
oid = ts_hex + rand_hex # 24 chars
# 解析 multipart/form-data
content_type = headers.get('content-type', '')
boundary = ''
if 'boundary=' in content_type:
boundary = content_type.split('boundary=')[1].split(';')[0].strip()
if boundary.startswith('"') and boundary.endswith('"'):
boundary = boundary[1:-1]
if not boundary:
return (400, {'error': 'invalid content-type, need multipart/form-data'})
# 从 body 解析文件
boundary_bytes = boundary.encode('utf-8')
parts = body.split(b'--' + boundary_bytes)
filename = 'unnamed'
file_data = None
task_id = ''
content_type_header_detected = ''
for part in parts:
if b'Content-Disposition' not in part:
continue
header_end = part.find(b'\r\n\r\n')
if header_end == -1:
continue
header_part = part[:header_end].decode('utf-8', errors='replace')
data_part = part[header_end+4:]
# 去掉尾部 \r\n--
if data_part.endswith(b'\r\n'):
data_part = data_part[:-2]
if data_part.endswith(b'--'):
data_part = data_part[:-2]
# 文件名
fn_match = re.search(r'filename="([^"]*)"', header_part)
if fn_match:
filename = fn_match.group(1)
# form字段名
name_match = re.search(r'name="([^"]*)"', header_part)
if name_match:
field_name = name_match.group(1)
if field_name == 'task_id':
task_id = data_part.decode('utf-8', errors='replace').strip()
continue
elif field_name == 'file':
file_data = data_part
# 检测 MIME
ct_match = re.search(r'Content-Type:\s*(\S+)', header_part, re.I)
if ct_match:
content_type_header_detected = ct_match.group(1)
if not file_data:
return (400, {'error': 'no file data in upload'})
if not task_id:
return (400, {'error': 'task_id required'})
filesize = len(file_data)
filename_safe = re.sub(r'[^\w\.\-\u4e00-\u9fff\(\)\[\]]', '_', filename)
if not filename_safe:
filename_safe = 'unnamed'
# 写临时文件
tmp_dir = '/tmp/chat_uploads'
os.makedirs(tmp_dir, exist_ok=True)
tmp_path = f"{tmp_dir}/{oid}_{filename_safe}"
with open(tmp_path, 'wb') as f:
f.write(file_data)
# OSS 路径
oss_key = f"attachments/{oid}/{filename_safe}"
oss_path = f"oss://rosicky-doc/{oss_key}"
ok, out = subprocess.getstatusoutput(f"ossutil cp -f {tmp_path} {oss_path}")
if ok != 0:
os.remove(tmp_path)
return (500, {'error': f'oss upload failed: {out[:200]}'})
# 清理临时文件
os.remove(tmp_path)
# 文件名加时间戳前缀: 0728_a.txt
date_prefix = time.strftime('%m%d')
timestamped_name = f"{date_prefix}_{filename}"
filename_safe = re.sub(r'[^\w\.\-\u4e00-\u9fff\(\)\[\]]', '_', timestamped_name)
if not filename_safe:
filename_safe = f"{date_prefix}_unnamed"
# 写入 DB
conn_att = sqlite3.connect(CONTENT_DB)
conn_att.execute(
"INSERT INTO attachments (id, task_id, filename, filesize, mime_type, oss_key, display_name, attach_type) VALUES (?,?,?,?,?,?,?,?)",
(oid, task_id, filename_safe, filesize, content_type_header_detected, oss_key, filename, 'file')
)
conn_att.commit()
conn_att.close()
print(f"[ATTACH] {filename_safe} ({filesize}B) -> https://rosicky-doc.oss-cn-shanghai.aliyuncs.com/{oss_key}", file=sys.stderr)
return (200, {'ok': True, 'id': oid, 'filename': filename_safe, 'filesize': filesize, 'url': f"https://rosicky-doc.oss-cn-shanghai.aliyuncs.com/{oss_key}"})
elif path.startswith('/chat/attachments') and method == 'GET':
"""查询附件列表"""
task_id = params.get('task_id', [''])[0]
limit = min(int(params.get('limit', ['50'])[0]), 100)
offset = int(params.get('offset', ['0'])[0])
try:
conn_att = sqlite3.connect(CONTENT_DB)
conn_att.row_factory = sqlite3.Row
if task_id:
rows = conn_att.execute(
"SELECT * FROM attachments WHERE task_id=? ORDER BY created_at DESC LIMIT ? OFFSET ?",
(task_id, limit, offset)
).fetchall()
total = conn_att.execute("SELECT COUNT(*) FROM attachments WHERE task_id=?", (task_id,)).fetchone()[0]
else:
rows = conn_att.execute(
"SELECT * FROM attachments ORDER BY created_at DESC LIMIT ? OFFSET ?",
(limit, offset)
).fetchall()
total = conn_att.execute("SELECT COUNT(*) FROM attachments").fetchone()[0]
conn_att.close()
result = []
for r in rows:
result.append({
'id': r['id'],
'task_id': r['task_id'],
'filename': r['filename'],
'filesize': r['filesize'],
'mime_type': r['mime_type'],
'created_at': r['created_at'],
'url': f"https://rosicky-doc.oss-cn-shanghai.aliyuncs.com/{r['oss_key']}",
})
return (200, {'attachments': result, 'total': total})
except Exception as e:
return (500, {'error': str(e)})
elif path.startswith('/chat/attachment/') and path.endswith('/download') and method == 'GET':
"""附件下载（返回 OSS 直链）"""
att_id = path.split('/')[3]
try:
conn_att = sqlite3.connect(CONTENT_DB)
row = conn_att.execute("SELECT * FROM attachments WHERE id=?", (att_id,)).fetchone()
conn_att.close()
if not row:
return (404, {'error': 'attachment not found'})
oss_url = f"https://rosicky-doc.oss-cn-shanghai.aliyuncs.com/{row[5]}"
return (200, {'url': oss_url, 'filename': row[2], 'filesize': row[3]})
except Exception as e:
return (500, {'error': str(e)})
elif path == '/chat/attachment/delete' and method == 'POST':
"""删除附件"""
try:
data = json.loads(body.decode('utf-8'))
att_id = data.get('id', '')
if not att_id:
return (400, {'error': 'id required'})
conn_att = sqlite3.connect(CONTENT_DB)
row = conn_att.execute("SELECT oss_key FROM attachments WHERE id=?", (att_id,)).fetchone()
if row:
subprocess.getstatusoutput(f"ossutil rm oss://rosicky-doc/{row[0]}")
conn_att.execute("DELETE FROM attachments WHERE id=?", (att_id,))
conn_att.commit()
conn_att.close()
return (200, {'ok': True})
except Exception as e:
return (400, {'error': str(e)})
elif path == '/chat/message/edit' and method == 'POST':
# 编辑消息
try:
data = json.loads(body.decode('utf-8'))
msg_id = data.get('id')
new_content = data.get('content', '').strip()
if not msg_id or not new_content:
return (400, {'error': 'id and content required'})
ok = update_chat_message(msg_id, new_content)
if ok:
msg = get_chat_message_by_id(msg_id)
return (200, {'success': True, 'message': msg})
return (404, {'error': 'message not found or not editable'})
except Exception as e:
return (400, {'error': str(e)})
elif path == '/chat/message/delete' and method == 'POST':
try:
data = json.loads(body.decode('utf-8'))
msg_id = data.get('id')
if not msg_id:
return (400, {'error': 'id required'})
ok = delete_chat_message(msg_id)
return (200, {'success': ok})
except Exception as e:
return (400, {'error': str(e)})
elif path.startswith('/chat/search'):
task_id = params.get('task_id', [''])[0]
keyword = params.get('q', [''])[0]
limit = min(int(params.get('limit', ['20'])[0]), 50)
if not task_id or not keyword:
return (200, {'messages': [], 'count': 0})
result = search_chat_messages(task_id, keyword, limit)
return (200, result)
elif path.startswith('/skills'):
try:
conn = sqlite3.connect(CONTENT_DB)
conn.row_factory = sqlite3.Row
# 技能列表
skills = conn.execute(
"SELECT s.name, s.category, s.description, s.file_path FROM skills s ORDER BY s.category, s.name"
).fetchall()
# 技能→task 映射
skill_tasks_map = {}
st_rows = conn.execute(
"SELECT st.skill_name, st.task_id, t.name as task_name FROM skill_tasks st LEFT JOIN tasks t ON st.task_id = t.id"
).fetchall()
for row in st_rows:
sn = row['skill_name']
if sn not in skill_tasks_map:
skill_tasks_map[sn] = []
skill_tasks_map[sn].append({
'task_id': row['task_id'],
'task_name': row['task_name'] or row['task_id'],
})
conn.close()
result = []
for s in skills:
name = s['name']
result.append({
'name': name,
'category': s['category'] or '',
'description': s['description'] or '',
'file_path': s['file_path'],
'tasks': skill_tasks_map.get(name, []),
})
return (200, result)
except Exception as e:
return (500, {'error': str(e)})
elif path == '/skill-tasks' and method == 'POST':
try:
data = json.loads(body.decode('utf-8'))
skill_name = data.get('skill_name', '')
task_ids = data.get('task_ids', [])
if not skill_name:
return (400, {'error': 'skill_name required'})
conn = sqlite3.connect(CONTENT_DB)
conn.execute("DELETE FROM skill_tasks WHERE skill_name=?", (skill_name,))
for tid in task_ids:
if tid:
conn.execute("INSERT OR IGNORE INTO skill_tasks (skill_name, task_id) VALUES (?,?)", (skill_name, tid))
conn.commit()
conn.close()
return (200, {'ok': True})
except Exception as e:
return (400, {'error': str(e)})
elif path.startswith('/skill-content'):
try:
name = params.get('name', [None])[0]
if not name:
return (400, {'error': 'name required'})
conn = sqlite3.connect(CONTENT_DB)
row = conn.execute("SELECT file_path FROM skills WHERE name=?", (name,)).fetchone()
conn.close()
if not row:
return (404, {'error': 'skill not found'})
fp = row[0]
skill_path = os.path.join(os.path.expanduser('~/.hermes/skills'), fp)
if not os.path.exists(skill_path):
# Try SKILL.md variants
alt = skill_path
if not alt.endswith('.md'):
alt = os.path.join(skill_path, 'SKILL.md')
if not os.path.exists(alt):
alt2 = os.path.join(os.path.dirname(skill_path), 'SKILL.md')
if os.path.exists(alt2):
alt = alt2
else:
return (404, {'error': 'skill file not found', 'path': skill_path})
skill_path = alt
with open(skill_path, 'r', encoding='utf-8') as f:
content = f.read()
return (200, {'name': name, 'content': content, 'file_path': fp})
except Exception as e:
return (500, {'error': str(e)})
elif path.startswith('/doc-content'):
try:
doc_id = params.get('doc_id', [None])[0]
if not doc_id:
return (400, {'error': 'doc_id required'})
# Fetch from OSS
import subprocess
oss_key = f"oss://rosicky-doc/contents/{doc_id}"
r = subprocess.run(["ossutil", "cat", oss_key], stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=30)
content = r.stdout.decode('utf-8', errors='replace')
if not content.strip() and r.returncode != 0:
# Try with .md suffix
oss_key_md = f"oss://rosicky-doc/contents/{doc_id}.md"
r = subprocess.run(["ossutil", "cat", oss_key_md], stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=30)
content = r.stdout.decode('utf-8', errors='replace')
return (200, {'doc_id': doc_id, 'content': content})
except Exception as e:
return (500, {'error': str(e)})
elif path == '/chat/upload' and method == 'POST':
"""上传附件 — 接受 multipart/form-data，保存到 /var/www/webchat/uploads/"""
UPLOAD_DIR = '/var/www/webchat/uploads'
MAX_FILE_SIZE = 10 * 1024 * 1024 # 10MB
try:
content_type = headers.get('content-type', '')
if 'multipart/form-data' not in content_type:
return (400, {'error': 'expected multipart/form-data'})
# 提取 boundary
m = re.search(r'boundary=([^;]+)', content_type)
if not m:
return (400, {'error': 'boundary not found'})
boundary = m.group(1).strip().strip('"').encode()
# 解析 multipart body
body_bytes = body if isinstance(body, bytes) else body.encode('utf-8')
parts = body_bytes.split(b'--' + boundary)
file_data = None
filename = None
for part in parts:
if part.startswith(b'\r\n') and b'Content-Disposition' in part[:500]:
# 分离 header 和 body
header_end = part.find(b'\r\n\r\n')
if header_end == -1:
continue
header_section = part[:header_end].decode('utf-8', errors='replace')
data_start = header_end + 4
# 去掉尾部 \r\n--\r\n
file_bytes = part[data_start:]
if file_bytes.endswith(b'\r\n'):
file_bytes = file_bytes[:-2]
# 提取文件名
fn_match = re.search(r'filename="([^"]*)"', header_section)
if fn_match:
filename = fn_match.group(1)
file_data = file_bytes
if not file_data or not filename:
return (400, {'error': 'no file uploaded'})
if len(file_data) > MAX_FILE_SIZE:
return (400, {'error': f'file too large (max {MAX_FILE_SIZE//1024//1024}MB)'})
# 安全处理文件名
safe_name = re.sub(r'[^\w\.\-]', '_', filename)
import uuid
unique_name = f"{uuid.uuid4().hex[:8]}_{safe_name}"
file_path = os.path.join(UPLOAD_DIR, unique_name)
with open(file_path, 'wb') as f:
f.write(file_data)
url = f"/uploads/{unique_name}"
print(f"[UPLOAD] {filename} -> {file_path} ({len(file_data)} bytes)", file=sys.stderr)
return (200, {
'ok': True,
'url': url,
'filename': filename,
'size': len(file_data),
})
except Exception as e:
print(f"[UPLOAD ERROR] {e}", file=sys.stderr)
return (500, {'error': str(e)})
elif path == '/wiki/create' and method == 'POST':
"""创建 wiki 文档 — 写 OSS + 写 content.db，供微服务/主 agent 统一调用"""
try:
data = json.loads(body.decode('utf-8'))
title = (data.get('title') or '').strip()
content = (data.get('content') or '').strip()
folder_id = data.get('folder_id') or ''
task_id = data.get('task_id') or ''
if not title or not content:
return (400, {'error': 'title and content required'})
# 生成 doc_id（同现有模式：timestamp_based）
import random, time
_EPOCH = 1700000000000
ts = int(time.time() * 1000) - _EPOCH
doc_id = str((ts << 10) | random.randint(0, 1023))
# 写临时文件 → ossutil cp
tmp_path = f'/tmp/wiki_create_{doc_id}.md'
with open(tmp_path, 'w', encoding='utf-8') as f:
f.write(content)
import subprocess
r = subprocess.run(
['ossutil', 'cp', tmp_path, f'oss://rosicky-doc/contents/{doc_id}'],
stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=30
)
os.unlink(tmp_path) # 删临时文件
if r.returncode != 0:
return (500, {'error': f'OSS upload failed: {r.stderr.decode()[:200]}'})
# 写 content.db
conn = sqlite3.connect(CONTENT_DB)
if folder_id:
# 验证 folder 存在
row = conn.execute("SELECT id FROM folders WHERE id=?", (folder_id,)).fetchone()
if not row:
folder_id = ''
if not folder_id:
# 默认放到「未归类」或根目录
row = conn.execute("SELECT id FROM folders WHERE name='未归类' LIMIT 1").fetchone()
folder_id = row[0] if row else '-1'
conn.execute(
"INSERT INTO documents (id, title, type, folder_id) VALUES (?, ?, 1, ?)",
(doc_id, title, folder_id)
)
# 关联 task（可选）
if task_id:
conn.execute("INSERT OR IGNORE INTO doc_tasks (doc_id, task_id) VALUES (?, ?)", (doc_id, task_id))
conn.commit()
conn.close()
url = f"http://doc.rosicky.top/view?doc={doc_id}"
return (200, {
'ok': True,
'doc_id': doc_id,
'title': title,
'url': url
})
except Exception as e:
return (500, {'error': str(e)})
# ── 表单 API ──
elif path == '/api/forms' and method == 'GET':
task_id = params.get('task_id', [''])[0]
try:
conn = sqlite3.connect(CONTENT_DB)
conn.row_factory = sqlite3.Row
if task_id:
rows = conn.execute(
"SELECT id, task_id, title, schema_json, created_at, updated_at FROM form_schemas WHERE task_id=? ORDER BY created_at DESC",
(task_id,)
).fetchall()
else:
rows = conn.execute(
"SELECT id, task_id, title, schema_json, created_at, updated_at FROM form_schemas ORDER BY created_at DESC"
).fetchall()
forms = []
for r in rows:
f = dict(r)
f['schema_json'] = json.loads(f['schema_json'])
forms.append(f)
conn.close()
return (200, {'forms': forms})
except Exception as e:
return (500, {'error': str(e)})
elif path == '/api/forms' and method == 'POST':
try:
body_data = json.loads(body) if body else {}
task_id = body_data.get('task_id', '')
title = body_data.get('title', '')
schema = body_data.get('schema', {})
if not task_id or not title or not schema:
return (400, {'error': 'task_id, title, schema required'})
import uuid, time
form_id = f"fm_{int(time.time())}_{uuid.uuid4().hex[:8]}"
conn = sqlite3.connect(CONTENT_DB)
conn.execute(
"INSERT INTO form_schemas (id, task_id, title, schema_json) VALUES (?, ?, ?, ?)",
(form_id, task_id, title, json.dumps(schema, ensure_ascii=False))
)
# 同时创建空值记录
val_id = f"fv_{int(time.time())}_{uuid.uuid4().hex[:8]}"
conn.execute(
"INSERT INTO form_values (id, form_id, values_json) VALUES (?, ?, '{}')",
(val_id, form_id)
)
conn.commit()
conn.close()
return (200, {'ok': True, 'form_id': form_id, 'value_id': val_id})
except Exception as e:
return (500, {'error': str(e)})
elif path.startswith('/api/forms/') and path.endswith('/values') and method == 'GET':
form_id = path.split('/')[3]
try:
conn = sqlite3.connect(CONTENT_DB)
conn.row_factory = sqlite3.Row
row = conn.execute(
"SELECT id, form_id, values_json, created_at, updated_at FROM form_values WHERE form_id=?",
(form_id,)
).fetchone()
conn.close()
if not row:
return (404, {'error': 'no values for this form'})
vals = dict(row)
vals['values_json'] = json.loads(vals['values_json'])
return (200, vals)
except Exception as e:
return (500, {'error': str(e)})
elif path.startswith('/api/forms/') and path.endswith('/values') and method == 'POST':
form_id = path.split('/')[3]
try:
body_data = json.loads(body) if body else {}
values = body_data.get('values', {})
conn = sqlite3.connect(CONTENT_DB)
existing = conn.execute(
"SELECT id FROM form_values WHERE form_id=?", (form_id,)
).fetchone()
if existing:
conn.execute(
"UPDATE form_values SET values_json=?, updated_at=datetime('now') WHERE form_id=?",
(json.dumps(values, ensure_ascii=False), form_id)
)
else:
val_id = f"fv_{int(time.time())}_{uuid.uuid4().hex[:8]}"
conn.execute(
"INSERT INTO form_values (id, form_id, values_json) VALUES (?, ?, ?)",
(val_id, form_id, json.dumps(values, ensure_ascii=False))
)
conn.commit()
conn.close()
return (200, {'ok': True})
except Exception as e:
return (500, {'error': str(e)})
elif path.startswith('/api/forms/render/') and method == 'GET':
form_id = path.split('/')[-1]
try:
conn = sqlite3.connect(CONTENT_DB)
conn.row_factory = sqlite3.Row
srow = conn.execute(
"SELECT id, task_id, title, schema_json FROM form_schemas WHERE id=?", (form_id,)
).fetchone()
conn.close()
if not srow:
return (404, {'error': 'form not found'})
schema = json.loads(srow['schema_json'])
return (200, {
'form_id': srow['id'],
'title': srow['title'],
'schema': schema,
})
except Exception as e:
return (500, {'error': str(e)})
return (404, {'error': 'not found'})
async def handle_http(reader, writer):
try:
header_bytes = b''
while True:
line = await asyncio.wait_for(reader.readline(), timeout=10)
if not line:
break
header_bytes += line
if line == b'\r\n':
break
if not header_bytes:
writer.close()
return
lines = header_bytes.decode('utf-8', errors='replace').split('\r\n')
if not lines or not lines[0]:
writer.close()
return
first = lines[0].split(' ')
if len(first) < 2:
writer.close()
return
method = first[0]
path = first[1]
headers = {}
for line in lines[1:]:
if ':' in line:
k, v = line.split(':', 1)
headers[k.strip().lower()] = v.strip()
body = b''
if 'content-length' in headers:
cl = int(headers['content-length'])
if cl > 0:
body = await asyncio.wait_for(reader.readexactly(cl), timeout=10)
result_or_data = handle_api(method, path, headers, body)
status = result_or_data[0]
if isinstance(result_or_data, (list, tuple)) and len(result_or_data) == 3:
content, mime = result_or_data[1], result_or_data[2]
else:
data = result_or_data
if isinstance(data, (list, tuple)) and len(data) >= 2:
data = data[1]
content = json.dumps(data, ensure_ascii=False)
mime = 'application/json; charset=utf-8'
body_bytes = content.encode('utf-8') if isinstance(content, str) else content
resp = (
f'HTTP/1.1 {status} {"OK" if status == 200 else "Not Found"}\r\n'
f'Content-Type: {mime}\r\n'
f'Access-Control-Allow-Origin: *\r\n'
f'Content-Length: {len(body_bytes)}\r\n'
f'Connection: close\r\n'
f'\r\n'
).encode() + body_bytes
writer.write(resp)
await writer.drain()
except Exception as e:
print(f"[HTTP ERROR] {e}", file=sys.stderr)
import traceback
traceback.print_exc()
finally:
writer.close()
# ═══════════════════════════════════════════
# 启动
# ═══════════════════════════════════════════
async def main():
from websockets.asyncio.server import serve as ws_serve
# 启动时获取 Dashboard token
token = _fetch_dashboard_token()
if token:
HermesRpcClient._token_override = token
print(f"[启动] Dashboard token 获取成功", file=sys.stderr)
else:
print(f"[启动] ⚠️ 未能获取 Dashboard token，请在 dashboard 启动后再启动本服务", file=sys.stderr)
print(f"\n{'='*50}")
print(f" Hermes WebSocket 聊天后端 v4 — JSON-RPC")
print(f" WS: ws://{HOST}:{WS_PORT}/")
print(f" HTTP: http://{HOST}:{HTTP_PORT}/chat/...")
print(f" RPC: → {HERMES_DASHBOARD_WS} (Hermes Dashboard)")
print(f" API: /chat/tasks, /chat/history, /chat/tree")
print(f"{'='*50}\n")
async with ws_serve(handle_ws, HOST, WS_PORT):
http_server = await asyncio.start_server(handle_http, HOST, HTTP_PORT)
await asyncio.gather(
http_server.serve_forever(),
asyncio.Future()
)
if __name__ == '__main__':
asyncio.run(main())