#!/usr/bin/env python3 """ XMPP Agent Core — 统一版 ========================= Single bot core for all agents. Supports --agent xxm|mohe|zhiwei|xiaoguo. Usage: python xmpp_agent_core.py --agent xxm # xxm, uses chat_bridge python xmpp_agent_core.py --agent mohe # mohe, uses Hermes API python xmpp_agent_core.py --agent zhiwei # zhiwei, uses Hermes API python xmpp_agent_core.py --agent xiaoguo # xiaoguo, uses Hermes API Shares: PID lock, reconnect, MUC join, dedup, batching, coordinator protocol (GRANT/REVOKE), HTTP bridge. Differs only in LLM calling method (chat_bridge vs Hermes API). """ import os, sys, time, threading, asyncio, logging, json, re, ssl import urllib.request, http.server, urllib.parse # ── Windows selector loop (slixmpp needs it on Windows) ── if sys.platform == "win32": asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy()) # ── PATH: allow imports from gateway/scripts/ (proc_guard, chat_bridge) ── _GATEWAY_SCRIPTS = os.path.join(os.path.dirname(os.path.abspath(__file__)), "gateway", "scripts") sys.path.insert(0, _GATEWAY_SCRIPTS) # ═══════════════════════════════════════════════════════════════ # AGENTS Configuration # ═══════════════════════════════════════════════════════════════ AGENTS = { "mohe": { "jid": "mohe@yoin.fun", "password": "hermes123", "nick": "mohe", "name_cn": "莫荷", "http_port": 5804, "gateway": "http://localhost:8642/v1/chat/completions", "session_id": "xmpp-mohe-v2", "server": "127.0.0.1", "port": 5222, "muc_rooms": ["coregroup@conference.yoin.fun"], "mention": "@mohe/@莫荷", }, "zhiwei": { "jid": "zhiwei@yoin.fun", "password": "hermes123", "nick": "zhiwei", "name_cn": "知微", "http_port": 5805, "gateway": "http://localhost:8643/v1/chat/completions", "session_id": "xmpp-zhiwei", "server": "127.0.0.1", "port": 5222, "muc_rooms": ["coregroup@conference.yoin.fun"], "mention": "@zhiwei/@知微", }, "xiaoguo": { "jid": "xiaoguo@yoin.fun", "password": "hermes123", "nick": "xiaoguo", "name_cn": "小果", "http_port": 5806, "gateway": "http://localhost:8645/v1/chat/completions", "session_id": "xmpp-xiaoguo", "kanban_session_id": "xmpp-xiaoguo-kanban", "server": "127.0.0.1", "port": 5222, "muc_rooms": ["coregroup@conference.yoin.fun"], "mention": "@xiaoguo/@小果", }, "xxm": { "jid": "xxm@yoin.fun", "password": "hermes123", "nick": "xxm", "name_cn": "笑笑", # NOTE: 5802 was historically used but is now occupied by wechat-hermes-gateway's # xmpp_bot.py (independent service). Switched to 5807 to avoid dual-listener conflict # that caused ~50% of /easytier and /rdp requests to 400. See AGENTS.md / rdp.json. "http_port": 5807, "bridge_api_key": "xxm_bridge_8f3a2c", "bridge": "chat_bridge", # use local chat_bridge instead of Hermes API "session_id": "ses_xxm_xmpp", "kanban_session_id": "xmpp-xxm-kanban", "server": "192.168.1.246", # LAN direct connect "port": 5222, "muc_rooms": [ "coregroup@conference.yoin.fun", "jujidina@conference.yoin.fun", ], "mention": "@xxm/@笑笑", }, } # ── Agent selection ── _agent_name = "mohe" if "--agent" in sys.argv: idx = sys.argv.index("--agent") if idx + 1 < len(sys.argv): _agent_name = sys.argv[idx + 1] cfg = AGENTS.get(_agent_name, AGENTS["mohe"]) # ═══════════════════════════════════════════════════════════════ # PID Lock — prevent duplicate instances # ═══════════════════════════════════════════════════════════════ from proc_guard import guard as _proc_guard _lock = _proc_guard(f"xmpp_bot_{_agent_name}") if not _lock.ok: print(_lock.message, flush=True) sys.exit(1) # ═══════════════════════════════════════════════════════════════ # Logging # ═══════════════════════════════════════════════════════════════ _LOG_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "gateway", "logs") os.makedirs(_LOG_DIR, exist_ok=True) _LOG_FILE = os.path.join(_LOG_DIR, f"xmpp_{_agent_name}.log") _START_TIME = time.time() def log(m: str): with open(_LOG_FILE, "a", encoding="utf-8") as f: f.write(f"{time.strftime('%H:%M:%S')} {m}\n") logging.basicConfig(level=logging.INFO, format='%(asctime)s %(message)s') # ═══════════════════════════════════════════════════════════════ # LLM Bridge Init — abstracted per agent # ═══════════════════════════════════════════════════════════════ _IS_CHAT_BRIDGE = cfg.get("bridge") == "chat_bridge" _router = None # set only for chat_bridge (xxm) # ── Kanban session support ── _CALL_SEQ = 0 _KANBAN_SESSION_ID = cfg.get("kanban_session_id", None) if _IS_CHAT_BRIDGE: from chat_bridge import SessionBridge from session_router import SessionRouter _bridge = SessionBridge(session_id=cfg["session_id"]) _router = SessionRouter(bridge=_bridge, default_session=cfg["session_id"]) # Kanban-dedicated bridge + router for separate session _kanban_sid = _KANBAN_SESSION_ID or f"{cfg['session_id']}-kanban" _kanban_bridge = SessionBridge(session_id=_kanban_sid) _kanban_router = SessionRouter(bridge=_kanban_bridge, default_session=_kanban_sid) log(f"LLM: chat_bridge (session={cfg['session_id']})") log(f"Kanban: chat_bridge (session={_kanban_sid})") else: _opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) log(f"LLM: Hermes API ({cfg['gateway']})") def _call_llm(content: str, sender: str, is_group: bool = False, session_id: str | None = None) -> str: """Abstract LLM call. Returns raw response text (or empty string). If session_id provided and differs from default, route to kanban handler.""" if _IS_CHAT_BRIDGE: if session_id and session_id != cfg["session_id"]: # Prepends kanban-handler instructions so LLM knows how to handle kanban_content = ( "【看板处理协议】\n" "收到卡片后三步判断：\n" " A) 任务明确 → 直接执行 → 评论结果 + 更新状态\n" " B) 信息不足 → 评论提问 + 设为 blocked\n" " C) 不是我的活 → 评论说明 + 转派（如果能判断）\n" "\n" "汇报规则：\n" " - 任务完成 → 简短 DM 给老莫摘要\n" " - 评论/状态变更 → 不汇报\n" " - 追问/转派 → 不汇报\n" "\n" "可用 API：\n" " curl http://192.168.1.246:5803/api/kanban/t_xxx 查看卡片详情\n" " 更新操作走 Kanban Dashboard UI\n" "\n" "卡片上下文不够？→ 用 session_search 查历史\n" f"---\n{content}" ) return _kanban_router.route("xmpp", sender, kanban_content) or "" return _router.route("xmpp", sender, content) or "" else: return _call_hermes_api(content, session_id) def _call_hermes_api(content: str, session_id: str | None = None) -> str: """POST to Hermes API, return response text or empty string.""" target_sid = session_id or cfg["session_id"] try: payload = json.dumps({ "model": "hermes-agent", "messages": [{"role": "user", "content": content}] }).encode() req = urllib.request.Request(cfg["gateway"], data=payload, method="POST") req.add_header("Content-Type", "application/json") req.add_header("Authorization", "Bearer hermes123") req.add_header("X-Hermes-Session-Id", target_sid) result = _opener.open(req, timeout=600) data = json.loads(result.read()) reply = data.get("choices", [{}])[0].get("message", {}).get("content", "") return reply.strip() except Exception as e: log(f"!!! Hermes API error: {e}") return "" # ═══════════════════════════════════════════════════════════════ # EasyTier control (Windows) # ═══════════════════════════════════════════════════════════════ _EASYTIER_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), "tools", "easytier") _EASYTIER_CORE = os.path.join(_EASYTIER_DIR, "easytier-core.exe") _EASYTIER_PID_FILE = os.path.join(_EASYTIER_DIR, "easytier.pid") _EASYTIER_NET = "--network-name mynet --network-secret ce75d0a5" _EASYTIER_RELAY = "--peers tcp://47.115.32.206:11010" _EASYTIER_IP = "--ipv4 10.144.144.3" _EASYTIER_FLAGS = "--disable-encryption" # TEMP: match relay+246; TODO remove after relay/246 also remove it # ============================================================ # RDP Toggle - remote desktop via SSH reverse tunnel (port 8080) # ============================================================ _RDP_SSH_HOST = 'root@47.115.32.206' _RDP_SSH_PORT = '8080' _RDP_PID_FILE = os.path.join(os.path.dirname(__file__), 'gateway', 'scripts', 'rdp_tunnel.pid') def _rdp_tunnel_pid(): try: if os.path.exists(_RDP_PID_FILE): with open(_RDP_PID_FILE) as f: return int(f.read().strip()) except Exception: pass return None def _rdp_health_check(): """Quick X.224 handshake test against localhost:3389. Returns True if RDP responds.""" import socket as _sk try: _s = _sk.create_connection(('127.0.0.1', 3389), timeout=5) # X.224 Connection Request + RDP Negotiation Request (SSL|HYBRID|RDSTLS) # TPKT len=0x13(19) 必须等于实际包长，否则服务器正确 RST（2026-07-22 误诊教训） _s.sendall(bytes.fromhex('030000130ee00000000000010008000b000000')) _s.settimeout(5) _resp = _s.recv(1024) _s.close() if len(_resp) >= 6 and _resp[5] in (0xd0, 0x03): return True except Exception: pass return False def _rdp_kill_huorong(): """Try to temporarily disable Huorong network protection that may block RDP.""" import subprocess as _sp try: # Unload sysdiag filter driver if present _sp.run(['fltmc', 'unload', 'sysdiag'], capture_output=True, timeout=5) log('RDP: Huorong sysdiag filter unloaded') except Exception: pass # Also try to stop Huorong services (may be protected, but worth a try) try: _sp.run(['taskkill', '/f', '/im', 'HipsDaemon.exe'], capture_output=True, timeout=5) except Exception: pass def _rdp_health_restore(): """Full RDP health restore: restart TermService + add firewall rule + verify.""" import subprocess as _sp, time as _t log('RDP: health check failed — attempting restore...') # 1. Add explicit WFP allow rule for port 3389 try: _sp.run(['netsh', 'advfirewall', 'firewall', 'add', 'rule', 'name=RDP-3389-TCP-AutoHeal', 'dir=in', 'protocol=tcp', 'localport=3389', 'action=allow', 'profile=any'], capture_output=True, timeout=10) log('RDP: added netsh allow rule for 3389') except Exception: pass # 2. Try to unload Huorong filter _rdp_kill_huorong() # 3. Full TermService restart _t.sleep(1) try: _sp.run(['net', 'stop', 'TermService', '/y'], capture_output=True, timeout=30) except Exception: _sp.run(['taskkill', '/f', '/im', 'svchost*'], capture_output=True, timeout=5) _t.sleep(3) try: _sp.run(['net', 'start', 'TermService'], capture_output=True, timeout=30) except Exception: pass _t.sleep(5) # 4. Verify ok = _rdp_health_check() log(f'RDP: restore result: {"OK" if ok else "STILL FAILING"}') return ok def _rdp_enable(): """Robust RDP enable — async step-machine. Writes live progress to rdp_progress.json + structured lines to rdp_enable.log, returns immediately with 'started'; dashboard polls /rdp action=progress for live steps.""" import threading as _th, json as _j try: if os.path.exists(_RDP_PROGRESS_FILE): with open(_RDP_PROGRESS_FILE, encoding='utf-8') as f: cur = _j.load(f) if cur.get('state') == 'running': return True, 'RDP enable 已在进行中（查看进度）' except Exception: pass _rdp_write_progress('running', 'init', _RDP_STEP_MSG['init'], []) _rdp_log('init', 'start', 'RDP enable requested') _th.Thread(target=_rdp_enable_run, daemon=True).start() return True, 'RDP enable 已启动（后台执行中，可看实时进度）' def _rdp_enable_run(): """The actual step-machine, runs in a daemon thread. Each step is recorded to rdp_progress.json (live) and rdp_enable.log (structured, for later debug).""" import subprocess as _sp, winreg as _wr, time as _t steps = [] def mark(name, status, detail=''): steps.append({'name': name, 'status': status, 'detail': detail, 'ts': _rdp_now()}) _rdp_write_progress('running', name, _RDP_STEP_MSG.get(name, name), steps, detail) _rdp_log(name, status, detail) try: # Step 1: 开注册表（开RDP + 隧道场景统一关NLA——SSH已加密，NLA的CredSSP预认证 # 正是"卡正在配置远程电脑"的根源：连接到达TermService但NLA协商挂起、建不了会话） try: k = _wr.OpenKey(_wr.HKEY_LOCAL_MACHINE, r'SYSTEM\CurrentControlSet\Control\Terminal Server', 0, _wr.KEY_SET_VALUE) _wr.SetValueEx(k, 'fDenyTSConnections', 0, _wr.REG_DWORD, 0) _wr.CloseKey(k) mark('registry', 'ok', 'fDenyTSConnections=0') except Exception as e: mark('registry', 'fail', str(e)) try: k2 = _wr.OpenKey(_wr.HKEY_LOCAL_MACHINE, r'SYSTEM\CurrentControlSet\Control\Terminal Server\WinStations\RDP-Tcp', 0, _wr.KEY_SET_VALUE) _wr.SetValueEx(k2, 'UserAuthentication', 0, _wr.REG_DWORD, 0) _wr.CloseKey(k2) mark('registry', 'ok', 'NLA已关闭(UserAuthentication=0，隧道场景)') except Exception as e: mark('registry', 'warn', '关NLA失败: ' + str(e)) # 隧道/多GPU/DisplayLink 环境统一禁用硬件图形适配器(软件渲染)： # DisplayLink USB 显示扩展坞/NVIDIA 多GPU 与 RDP WDDM 图形初始化冲突， # 是"卡正在配置远程电脑"(WDDM启用后挂起)的高频根因。软件渲染绕开它。 try: import subprocess as _sp2 _sp2.run(['powershell', '-NoProfile', '-Command', "New-Item -Path 'HKLM:\\SOFTWARE\\Policies\\Microsoft\\Windows NT\\Terminal Services' -Force | Out-Null; Set-ItemProperty -Path 'HKLM:\\SOFTWARE\\Policies\\Microsoft\\Windows NT\\Terminal Services' -Name 'bEnumerateHWBeforeSW' -Value 0 -Type DWord"], capture_output=True, timeout=10) mark('registry', 'ok', '已禁用硬件图形适配器(软件渲染，绕开DisplayLink/多GPU冲突)') except Exception as e: mark('registry', 'warn', '禁用硬件图形失败: ' + str(e)) # Step 2: 加入 RDP 用户组 try: _sp.run(['net', 'localgroup', 'Remote Desktop Users', 'hmo', '/add'], capture_output=True, timeout=8) mark('rdp_users', 'ok', 'hmo 已加入 Remote Desktop Users') except Exception as e: mark('rdp_users', 'warn', str(e)) # Step 3: 深度健康检查 hc = _rdp_deep_health_check() mark('health_check', 'ok' if hc['healthy'] else 'warn', 'x224=%s svc=%s listen=%s 卡死=%d nla=%s cert=%s' % ( hc['x224'], hc['termservice'], hc['port_listen'], hc['stale_sessions'], hc['nla'], hc['cert_ok'])) # Step 4: 自愈（如需） if not hc['healthy']: _rdp_log('restore', 'start', 'issues: ' + '; '.join(hc['issues'])) cleared = _rdp_kill_stale_rdp_sessions() if cleared: _rdp_log('restore', 'detail', '清理卡死会话: ' + ','.join(cleared)) _rdp_health_restore() hc2 = _rdp_deep_health_check() # NLA 卡死处理：连接能到但建不了会话且 NLA 开 → 临时关 NLA 再重启（隧道内网场景可逆） if not hc2['healthy'] and hc2.get('nla') == 1: _rdp_log('restore', 'detail', '仍异常且NLA开启 → 临时关闭NLA并重启TermService') _rdp_set_nla(False) _t.sleep(1) _sp.run(['net', 'stop', 'TermService', '/y'], capture_output=True, timeout=30) _t.sleep(2) _sp.run(['net', 'start', 'TermService'], capture_output=True, timeout=30) _t.sleep(3) mark('restore', 'ok', '自愈完成: ' + ('; '.join(hc['issues']) or 'no-op')) else: mark('restore', 'skip', '健康，无需自愈') # Step 5: 建立隧道 _rdp_kill_tunnel() _t.sleep(4) # 等阿里云端 8080 转发随旧 ssh 断开而释放，避免新隧道 forwarding failed _existing = _rdp_find_tunnel_process() if _existing: _rdp_adopt_tunnel(_existing) mark('tunnel', 'ok', '收养已有隧道 PID %d' % _existing) else: ok, msg = _rdp_start_tunnel() mark('tunnel', 'ok' if ok else 'fail', msg) if not ok: _rdp_write_progress('failed', 'tunnel', '隧道建立失败', steps, msg) _rdp_log('done', 'failed', msg) return # Step 6: 端到端验证 _t.sleep(2) if _rdp_verify_endtoend(): mark('verify', 'ok', '端到端握手通过 (47.115.32.206:8080)') _rdp_write_progress('done', 'done', '✓ RDP 已就绪，可连接 47.115.32.206:8080', steps, '') _rdp_log('done', 'ok', 'RDP fully enabled and verified') else: mark('verify', 'warn', '端到端验证未通过（隧道通但RDP握手失败，可能需重试）') _rdp_write_progress('done', 'done', '隧道已建立，端到端验证未通过，请稍后重试连接', steps, '') _rdp_log('done', 'warn', 'tunnel up but end-to-end verify failed') except Exception as e: mark('done', 'fail', str(e)) _rdp_write_progress('failed', 'done', '启动失败: ' + str(e), steps, str(e)) _rdp_log('done', 'failed', str(e)) def _rdp_disable(): import subprocess as _sp, winreg as _wr _rdp_kill_tunnel() try: k = _wr.OpenKey(_wr.HKEY_LOCAL_MACHINE, r'SYSTEM\CurrentControlSet\Control\Terminal Server', 0, _wr.KEY_SET_VALUE) _wr.SetValueEx(k, 'fDenyTSConnections', 0, _wr.REG_DWORD, 1) _wr.CloseKey(k) log('RDP access disabled') return True, 'RDP access disabled' except Exception as e: return False, 'Failed to disable RDP: ' + str(e) def _rdp_kill_tunnel(): import subprocess as _sp pid = _rdp_tunnel_pid() if pid: try: _sp.run(['taskkill', '/f', '/pid', str(pid)], capture_output=True, timeout=5) log('Killed RDP tunnel (PID ' + str(pid) + ')') except Exception: pass try: if os.path.exists(_RDP_PID_FILE): os.remove(_RDP_PID_FILE) except Exception: pass def _rdp_find_tunnel_process(): """Find a running ssh.exe hosting the -R 8080:localhost:3389 reverse tunnel. Returns PID or None. Used to adopt tunnels started outside _rdp_enable (e.g. by a crashed/restarted bot that left a detached ssh behind).""" import subprocess as _sp try: # 纯单引号写法：避免 powershell -Command 下双引号转义失效 ps_cmd = ("Get-CimInstance Win32_Process | " "Where-Object { $_.Name -eq 'ssh.exe' -and $_.CommandLine -like '*8080:localhost:3389*' } | " "Select-Object -ExpandProperty ProcessId") r = _sp.run(['powershell', '-NoProfile', '-Command', ps_cmd], capture_output=True, text=True, timeout=15) for line in r.stdout.splitlines(): line = line.strip() if line.isdigit(): return int(line) except Exception: pass return None def _rdp_adopt_tunnel(pid): """Write pid file for an externally-started tunnel so status tracking works.""" try: with open(_RDP_PID_FILE, 'w') as f: f.write(str(pid)) log(f'RDP: adopted existing tunnel (PID {pid})') return True except Exception: return False def _rdp_status(): import subprocess as _sp, winreg as _wr tunnel_on = False pid = _rdp_tunnel_pid() if pid: try: r = _sp.run(['tasklist', '/fi', 'PID eq ' + str(pid), '/fi', 'imagename eq ssh.exe'], capture_output=True, text=True, timeout=5) tunnel_on = 'ssh.exe' in r.stdout except Exception: pass # pid file missing or process dead → look for an actual tunnel process and adopt it if not tunnel_on: found = _rdp_find_tunnel_process() if found: _rdp_adopt_tunnel(found) tunnel_on = True rdp_on = False try: k = _wr.OpenKey(_wr.HKEY_LOCAL_MACHINE, r'SYSTEM\CurrentControlSet\Control\Terminal Server', 0, _wr.KEY_READ) v, _ = _wr.QueryValueEx(k, 'fDenyTSConnections') _wr.CloseKey(k) rdp_on = (v == 0) except Exception: pass return {'ok': True, 'tunnel_running': tunnel_on, 'rdp_enabled': rdp_on, 'rdp_port': 3389, 'tunnel_host': 'root@47.115.32.206', 'tunnel_port': 8080} # ============================================================ # RDP Robustness — step-machine + progress + structured log # 让 enable 按钮"点了就能用"：深度检查 + 自愈 + 端到端验证 + 实时进度 # ============================================================ _RDP_PROGRESS_FILE = os.path.join(os.path.dirname(__file__), 'gateway', 'scripts', 'rdp_progress.json') _RDP_ENABLE_LOG = os.path.join(_LOG_DIR, 'rdp_enable.log') _RDP_STEP_MSG = { 'init': '开始启动 RDP 远程桌面', 'registry': '开启远程桌面注册表', 'rdp_users': '加入远程桌面用户组', 'health_check': '深度健康检查（TermService/端口/卡死连接/NLA/证书）', 'restore': '检测到异常，正在自我修复', 'tunnel': '建立 SSH 反向隧道', 'verify': '端到端连通性验证', 'done': '完成', } def _rdp_now(): import time as _t return _t.strftime('%Y-%m-%d %H:%M:%S') def _rdp_log(step, status, detail=''): """Append a structured line to rdp_enable.log AND mirror to bot log.""" line = ("[%s] [%s] %s %s" % (_rdp_now(), step.upper(), status.upper(), detail)).rstrip() try: with open(_RDP_ENABLE_LOG, 'a', encoding='utf-8') as f: f.write(line + '\n') except Exception: pass log('RDP ' + line) def _rdp_write_progress(state, step, message, steps=None, detail=''): """Write rdp_progress.json for the dashboard to poll live steps.""" import json as _j prog = {'state': state, 'step': step, 'message': message, 'detail': detail, 'steps': steps or [], 'updated': _rdp_now()} try: with open(_RDP_PROGRESS_FILE, 'w', encoding='utf-8') as f: _j.dump(prog, f, ensure_ascii=False, indent=1) except Exception: pass def _rdp_progress(): """Return current enable progress (for /rdp action=progress).""" import json as _j try: if os.path.exists(_RDP_PROGRESS_FILE): with open(_RDP_PROGRESS_FILE, encoding='utf-8') as f: d = _j.load(f) d['ok'] = True return d except Exception: pass return {'ok': True, 'state': 'idle', 'step': '', 'message': '无进行中的任务', 'steps': [], 'detail': ''} def _rdp_read_enable_log(lines=80): """Tail the structured enable log (for /rdp action=enable_log).""" try: if os.path.exists(_RDP_ENABLE_LOG): with open(_RDP_ENABLE_LOG, encoding='utf-8', errors='replace') as f: data = f.readlines() return {'ok': True, 'lines': [l.rstrip('\n') for l in data[-lines:]]} except Exception as e: return {'ok': False, 'error': str(e), 'lines': []} return {'ok': True, 'lines': []} def _rdp_deep_health_check(): """Deep RDP health check — unlike _rdp_health_check (X.224/TCP only), this also checks TermService, 3389 listen, stale/close-wait sessions, NLA setting and cert validity, so it can detect "connects but stuck at configuring" (NLA stall / dead session) that the shallow check blindly passes.""" import subprocess as _sp res = {'x224': _rdp_health_check(), 'termservice': False, 'port_listen': False, 'stale_sessions': 0, 'nla': None, 'cert_ok': None, 'healthy': False, 'issues': []} try: r = _sp.run(['sc', 'query', 'TermService'], capture_output=True, text=True, timeout=8) res['termservice'] = 'RUNNING' in r.stdout except Exception: pass try: r = _sp.run(['powershell', '-NoProfile', '-Command', "if(Get-NetTCPConnection -LocalPort 3389 -State Listen -EA SilentlyContinue){'yes'}"], capture_output=True, text=True, timeout=12) res['port_listen'] = 'yes' in r.stdout except Exception: pass try: r = _sp.run(['powershell', '-NoProfile', '-Command', "@(Get-NetTCPConnection -LocalPort 3389 -EA SilentlyContinue | Where-Object {$_.State -in 'CloseWait','FinWait1','FinWait2','LastAck','TimeWait'}).Count"], capture_output=True, text=True, timeout=12) n = r.stdout.strip() res['stale_sessions'] = int(n) if n.isdigit() else 0 except Exception: pass try: import winreg as _wr k = _wr.OpenKey(_wr.HKEY_LOCAL_MACHINE, r'SYSTEM\CurrentControlSet\Control\Terminal Server\WinStations\RDP-Tcp', 0, _wr.KEY_READ) v, _ = _wr.QueryValueEx(k, 'UserAuthentication') _wr.CloseKey(k) res['nla'] = int(v) except Exception: pass try: r = _sp.run(['powershell', '-NoProfile', '-Command', "$c=Get-ChildItem 'Cert:\\LocalMachine\\Remote Desktop' -EA SilentlyContinue | Where-Object {$_.HasPrivateKey -and $_.NotAfter -gt (Get-Date)} | Select-Object -First 1; if($c){'ok'}else{'none'}"], capture_output=True, text=True, timeout=12) res['cert_ok'] = 'ok' in r.stdout except Exception: pass issues = [] if not res['termservice']: issues.append('TermService未运行') if not res['port_listen']: issues.append('3389未监听') if not res['x224']: issues.append('X.224握手失败') if res['stale_sessions'] > 0: issues.append('卡死连接x%d' % res['stale_sessions']) if res['cert_ok'] is False: issues.append('RDP证书异常') res['issues'] = issues res['healthy'] = bool(res['x224'] and res['termservice'] and res['port_listen'] and res['stale_sessions'] == 0 and res['cert_ok'] is not False) return res def _rdp_kill_stale_rdp_sessions(): """Log off half-dead/disconnected RDP sessions and clear close-wait conns that block new session setup (a common cause of 'stuck at configuring').""" import subprocess as _sp cleared = [] try: r = _sp.run(['qwinsta'], capture_output=True, text=True, timeout=8) for line in r.stdout.splitlines(): if 'rdp-tcp#' in line and ('Disc' in line or '断开' in line or 'Down' in line): parts = line.split() for p in parts: if p.isdigit() and int(p) > 0: _sp.run(['logoff', p], capture_output=True, timeout=5) cleared.append('session#' + p) break except Exception: pass return cleared def _rdp_set_nla(enabled): """Set NLA (UserAuthentication) on RDP-Tcp. enabled=True→1, False→0.""" import winreg as _wr try: k = _wr.OpenKey(_wr.HKEY_LOCAL_MACHINE, r'SYSTEM\CurrentControlSet\Control\Terminal Server\WinStations\RDP-Tcp', 0, _wr.KEY_SET_VALUE) _wr.SetValueEx(k, 'UserAuthentication', 0, _wr.REG_DWORD, 1 if enabled else 0) _wr.CloseKey(k) return True except Exception: return False def _rdp_start_tunnel(): """Start the SSH reverse tunnel (extracted from old _rdp_enable body). Retries when the Aliyun-side 8080 forwarding is still held by a just-killed previous tunnel ('remote port forwarding failed for listen port 8080') — that port needs a few seconds to be released after the old ssh dies.""" import subprocess as _sp, time as _t, os _retry_waits = [8, 12, 18, 25] # 递增等待：覆盖阿里云端 8080 转发随旧 ssh 断开而缓慢释放的时间 for _attempt in range(5): try: _ssh_key = None for _user_home in [os.path.expanduser('~'), r'C:\Users\hmo']: _candidate = os.path.join(_user_home, '.ssh', 'id_rsa') if os.path.isfile(_candidate): _ssh_key = _candidate break if not _ssh_key: return False, 'SSH 私钥 ~/.ssh/id_rsa 未找到' cmd = ['ssh.exe', '-i', _ssh_key, '-o', 'StrictHostKeyChecking=no', '-o', 'ServerAliveInterval=30', '-o', 'ExitOnForwardFailure=yes', '-o', 'IdentitiesOnly=yes', '-N', '-R', '0.0.0.0:8080:localhost:3389', 'root@47.115.32.206'] si = _sp.STARTUPINFO() si.dwFlags |= _sp.STARTF_USESHOWWINDOW _ssh_stderr_path = os.path.join(_LOG_DIR, 'rdp_tunnel_ssh.log') _ssh_err_fh = open(_ssh_stderr_path, 'a', encoding='utf-8') _ssh_err_fh.write('\n' + '=' * 60 + '\n' + _t.strftime('%Y-%m-%d %H:%M:%S') + ' RDP tunnel start (attempt %d)\n' % (_attempt + 1)) _ssh_err_fh.flush() _ssh_env = os.environ.copy() _ssh_env['HOME'] = os.path.expanduser('~') _ssh_env['USERPROFILE'] = os.path.expanduser('~') p = _sp.Popen(cmd, startupinfo=si, stdout=_sp.DEVNULL, stderr=_ssh_err_fh, env=_ssh_env, creationflags=_sp.DETACHED_PROCESS | _sp.CREATE_NEW_PROCESS_GROUP) _t.sleep(3) if p.poll() is not None: _ssh_err_fh.close() # ssh 立即退出：判断是否 8080 被旧隧道残留占用（可重试） _fwd_fail = False try: with open(_ssh_stderr_path, encoding='utf-8', errors='replace') as _ef: _tail = _ef.read()[-600:] _fwd_fail = ('forwarding failed' in _tail) or ('8080' in _tail and 'listen' in _tail) except Exception: pass if _fwd_fail and _attempt < 4: _wait = _retry_waits[_attempt] _rdp_log('tunnel', 'warn', '8080 被旧隧道残留占用，等待 %ds 后重试 (%d/5)' % (_wait, _attempt + 1)) _rdp_write_progress('running', 'tunnel', 'SSH 反向隧道', None, '8080 残留占用，等待 %ds 重试 (%d/5)' % (_wait, _attempt + 1)) _t.sleep(_wait) continue return False, 'SSH 隧道立即退出(code %s)，见 %s' % (p.poll(), _ssh_stderr_path) with open(_RDP_PID_FILE, 'w') as f: f.write(str(p.pid)) return True, '隧道已建立 PID %d' % p.pid except Exception as e: if _attempt < 4: _t.sleep(_retry_waits[_attempt]) continue return False, '隧道建立异常: ' + str(e) return False, 'SSH 隧道多次失败（8080 可能被旧隧道残留占用，请稍后重试）' def _rdp_verify_endtoend(): """After tunnel is up, verify full chain via public endpoint 47.115.32.206:8080 with an X.224 handshake (proves 8080→tunnel→3389→RDP all work).""" import socket as _sk try: _s = _sk.create_connection(('47.115.32.206', 8080), timeout=8) _s.sendall(bytes.fromhex('030000130ee00000000000010008000b000000')) _s.settimeout(8) _resp = _s.recv(1024) _s.close() if len(_resp) >= 6 and _resp[5] in (0xd0, 0x03): return True except Exception: pass return False # ── OpenCode Go Usage Monitor helpers ── # Reads cached aggregation from gateway/temp/usage_stats.json. # Triggers asynchronous collection by spawning usage_collector.py in a daemon thread. # Cache file path + collector script path: _USAGE_CACHE_FILE = os.path.join(os.path.dirname(__file__), 'gateway', 'temp', 'usage_stats.json') _USAGE_COLLECTOR_SCRIPT = os.path.join(_GATEWAY_SCRIPTS, 'usage_collector.py') _usage_collector_running = False # module-level lock flag def _now_iso(): """UTC ISO timestamp (avoids importing datetime everywhere we need this).""" from datetime import datetime, timezone return datetime.now(timezone.utc).isoformat(timespec='seconds') def _usage_read_cache(): """Return cached usage stats from temp/usage_stats.json, or empty payload if missing.""" if not os.path.isfile(_USAGE_CACHE_FILE): return {'ok': False, 'error': 'no cached data yet (collect never ran)', 'last_refresh_iso': None, 'accounts': []} try: with open(_USAGE_CACHE_FILE, 'r', encoding='utf-8') as f: data = json.load(f) # Don't pollute with no-stdout fields; just return the JSON object return data except Exception as e: return {'ok': False, 'error': f'failed to read cache: {e}', 'last_refresh_iso': None, 'accounts': []} def _usage_trigger_async(): """ Spawn a daemon thread that runs usage_collector.py via subprocess. Sets a module-level flag so concurrent calls don't overlap collections. The collector runs in its own process so it cannot crash the bot. Writes its result to gateway/temp/usage_stats.json (read by _usage_read_cache). """ global _usage_collector_running if _usage_collector_running: log('usage: collect_now already in-flight, skipping trigger') return _usage_collector_running = True def _runner(): global _usage_collector_running try: import subprocess as _sp log(f'usage: spawning usage_collector.py via {sys.executable}') # Run collector as one-shot --print so its stdout is captured but not displayed proc = _sp.run([sys.executable, _USAGE_COLLECTOR_SCRIPT], capture_output=True, text=True, timeout=120) log(f'usage: collector finished (rc={proc.returncode}, ' f'stdout_len={len(proc.stdout)}, stderr_len={len(proc.stderr)})') if proc.returncode != 0 and proc.stderr: log(f'usage_collector.py stderr:\n{proc.stderr[-1500:]}') except Exception as e: log(f'usage: collector thread crashed: {e}') finally: _usage_collector_running = False t = threading.Thread(target=_runner, name='usage_collector_runner', daemon=True) t.start() log('usage: collect_now background thread spawned') # ── Auto-collect timer (every 5 min, keeps cookies alive) ── _USAGE_AUTO_INTERVAL = 300 # 5 minutes def _start_usage_auto_timer(): """Start a daemon thread that auto-collects every 5 min (cookie keepalive + fresh data).""" def _loop(): import time as _time _time.sleep(10) # initial delay: let bot stabilize while True: try: _usage_trigger_async() except Exception as e: log(f'usage: auto-collect error: {e}') _time.sleep(_USAGE_AUTO_INTERVAL) t = threading.Thread(target=_loop, name='usage_auto_timer', daemon=True) t.start() log(f'usage: auto-timer started (interval={_USAGE_AUTO_INTERVAL}s)') def _start_easytier(): """Start EasyTier on Windows.""" import subprocess as _sp import time as _time if not os.path.exists(_EASYTIER_CORE): log(f"EasyTier binary not found: {_EASYTIER_CORE}") return # Kill ALL existing easytier-core.exe processes FIRST (prevents duplicate conflicts) _sp.run(["taskkill", "/f", "/im", "easytier-core.exe"], capture_output=True, timeout=5) _time.sleep(1) # Clean stale PID file try: if os.path.exists(_EASYTIER_PID_FILE): os.remove(_EASYTIER_PID_FILE) except Exception: pass # Start cmd = f'start /b "" "{_EASYTIER_CORE}" {_EASYTIER_NET} {_EASYTIER_RELAY} {_EASYTIER_IP} {_EASYTIER_FLAGS} --no-listener' try: _sp.run(cmd, shell=True, timeout=5) log("EasyTier start command issued") # Write PID file so future calls can track this instance _time.sleep(2) r = _sp.run(["tasklist", "/fi", "imagename eq easytier-core.exe", "/fo", "csv"], capture_output=True, text=True, timeout=5) import csv as _csv for row in _csv.reader(r.stdout.splitlines()): if len(row) >= 2 and row[0].strip('"') == "easytier-core.exe": pid = int(row[1]) with open(_EASYTIER_PID_FILE, "w") as f: f.write(str(pid)) log(f"EasyTier started (PID {pid})") break except Exception as e: log(f"EasyTier start error: {e}") def _stop_easytier(): """Stop EasyTier on Windows.""" import subprocess as _sp try: _sp.run(["taskkill", "/f", "/im", "easytier-core.exe"], capture_output=True, timeout=5) log("EasyTier stopped (taskkill)") except Exception as e: log(f"EasyTier stop error: {e}") # Clean up PID file try: if os.path.exists(_EASYTIER_PID_FILE): os.remove(_EASYTIER_PID_FILE) except Exception: pass def _check_easytier() -> bool: """Check if EasyTier is running on Windows.""" import subprocess as _sp try: r = _sp.run(["tasklist", "/fi", "imagename eq easytier-core.exe"], capture_output=True, text=True, timeout=5) return "easytier-core.exe" in r.stdout except Exception: return False # ═══════════════════════════════════════════════════════════════ # Message Dedup # ═══════════════════════════════════════════════════════════════ _DEDUP_CACHE: set[str] = set() _DEDUP_LOCK = threading.Lock() def _is_duplicate(msg_id: str) -> bool: if not msg_id: return False with _DEDUP_LOCK: if msg_id in _DEDUP_CACHE: return True _DEDUP_CACHE.add(msg_id) if len(_DEDUP_CACHE) > 100: _DEDUP_CACHE.clear() return False # ═══════════════════════════════════════════════════════════════ # Coordinator Protocol (shared across all agents) # ═══════════════════════════════════════════════════════════════ _COORDINATOR: str = "mohe" _GRANTED: str | None = None _REVOKED_UNTIL: float = 0.0 _SHUTUP_PATTERNS = ["闭嘴", "别说话", "安静", "shut", "stfu", "别说了", "停"] def _process_coordinator_signals(nickname: str, body: str) -> bool: """Parse coordinator/GRANT/REVOKE from incoming messages. Returns True if message was a control signal (consumed, no further processing).""" global _COORDINATOR, _GRANTED, _REVOKED_UNTIL # 1. hmo switches coordinator if nickname == 'hmo' and 'coordinator=' in body.lower(): for name in ('mohe', 'zhiwei', 'xxm'): if f'coordinator={name}' in body.lower(): _COORDINATOR = name _GRANTED = None log(f"Coordinator switched to {name} by hmo") return True # 2. GRANT signal (overrides REVOKE) gm = re.search(r'\[GRANT:(\w+)\]', body) if gm: _GRANTED = gm.group(1) _REVOKED_UNTIL = 0 log(f"GRANT: {_GRANTED}") return True # 3. REVOKE signal (5min auto-restore) rm = re.search(r'\[REVOKE:(\w+)\]', body) if rm and rm.group(1) == cfg["nick"]: _REVOKED_UNTIL = time.time() + 300 log(f"REVOKEd: {cfg['nick']} silenced for 5min") return True return False def _check_shutup(body: str) -> bool: """hmo says shut up → 5min silence.""" lower = body.lower().strip() for pat in _SHUTUP_PATTERNS: if pat.lower() in lower: _REVOKED_UNTIL = time.time() + 300 log(f"(shutup: '{pat}' → 5min silence)") return True return False def _process_llm_grant(response: str): """Parse GRANT signal from LLM's own response.""" global _GRANTED gm = re.search(r'\[GRANT:(\w+)\]', response) if gm: _GRANTED = gm.group(1) _REVOKED_UNTIL = 0 log(f"LLM GRANT: {_GRANTED}") # ═══════════════════════════════════════════════════════════════ # Response Extraction # ═══════════════════════════════════════════════════════════════ _SILENCE_PATTERNS = [ "保持沉默", "不应[该]?回复", "没有.*@.*我", "不是对[我我说]", "跟我无关", "我不用回复", "不该回复", "不参与", "不是我[应]?该[说回]", ] _EXEC_RE = re.compile(r"##exec:(.+?)##", re.DOTALL) _DELAY_RE = re.compile(r"##delay:?(\d+)?##") _DELAY_DEFAULT = 15 _EXEC_TIMEOUT = 60 def _strip_toolcall_xml(text: str) -> str: t = text t = re.sub(r']*>.*?(|$)', '', t, flags=re.DOTALL) t = re.sub(r'.*?(|$)', '', t, flags=re.DOTALL) t = re.sub(r']*>.*?(|$)', '', t, flags=re.DOTALL) t = re.sub(r'.*?(|$)', '', t, flags=re.DOTALL) return t.strip() def _extract_response(text: str) -> str | None: """Strip __SILENT__, reasoning blocks, natural language silence. Returns actual content to send, or None to stay silent.""" if not text: return None t = text.strip() if not t: return None t = _strip_toolcall_xml(t) # Natural language silence detection if not t.startswith("__SILENT__"): first = t.split("\n", 1)[0] for pat in _SILENCE_PATTERNS: if re.search(pat, first): return None return t # Has __SILENT__ prefix parts = t.split("\n", 1) if len(parts) < 2: return None rest = parts[1].strip() while True: m = re.match(r'^（[^）]*）\s*', rest) if m: rest = rest[m.end():] continue m = re.match(r'^\([^)]*\)\s*', rest) if m: rest = rest[m.end():] continue break return rest.strip() or None # ═══════════════════════════════════════════════════════════════ # Message Batching (3s debounce + serialized processing) # ═══════════════════════════════════════════════════════════════ _BATCH_WINDOW = 3.0 _BATCH_TIMEOUT = 300 _batch_entries: dict[str, list[str]] = {} _batch_timers: dict[str, threading.Timer] = {} _batch_processing: set[str] = set() _batch_pending: dict[str, list[str]] = {} _batch_lock = threading.Lock() _BOT_NICK = cfg["nick"] def _run_command(cmd: str) -> str: """Execute ##exec:command## shell command.""" log(f"(exec: {cmd[:120]})") try: import subprocess r = subprocess.run(cmd, shell=True, capture_output=True, timeout=_EXEC_TIMEOUT, text=True, encoding='utf-8', errors='replace') out = (r.stdout or "") + (r.stderr or "") out = out.strip() or f"(no output, exit={r.returncode})" log(f"(exec done: {len(out)} bytes, exit={r.returncode})") return out except subprocess.TimeoutExpired: log(f"(exec timeout >{_EXEC_TIMEOUT}s)") return "(命令超时)" except Exception as e: log(f"(exec error: {e})") return f"(命令执行失败: {e})" def _schedule_delayed(delay_sec: int, room: str): """Schedule ##delay:N## re-invocation.""" global _xmpp_ref import subprocess as _sp def _fire(): bot = _xmpp_ref if not bot: return try: prompt = "时间到，请根据最新的信息汇报结果。" raw = _call_llm(prompt, room, is_group=True) reply = _extract_response(raw) if reply: text = reply.strip() bot.send_message(mto=room, mbody=text, mtype='groupchat') log(f"-> [Delay][{room}]: {text[:80]}") except Exception as e: log(f"!! delay err: {e}") t = threading.Timer(delay_sec, _fire) t.daemon = True t.start() log(f"(delay +{delay_sec}s → {room})") def _batch_done(room: str): """Called when batch LLM finishes. Flush pending if any.""" with _batch_lock: _batch_processing.discard(room) pending = _batch_pending.pop(room, None) if pending: _batch_entries[room] = pending t = threading.Timer(0.1, _fire_batch, args=[room]) t.daemon = True t.start() _batch_timers[room] = t return log(f"[Batch][{room}] (idle)") def _fire_batch(room: str): """Collect batched entries and call LLM.""" with _batch_lock: entries = _batch_entries.pop(room, None) _batch_timers.pop(room, None) if not entries: return _batch_processing.add(room) combined = "\n".join(entries) def _handle(): timed_out = [False] def _timeout(): timed_out[0] = True log(f"[Batch][{room}] TIMEOUT ({_BATCH_TIMEOUT}s)") _batch_done(room) timer = threading.Timer(_BATCH_TIMEOUT, _timeout) timer.daemon = True timer.start() try: raw = _call_llm(combined, room, is_group=True) if not timed_out[0]: timer.cancel() _process_llm_reply(raw, room) else: log(f"[Batch][{room}] route returned after timeout, discarded") except Exception as e: log(f"!!! BATCH: {e}") if not timed_out[0]: timer.cancel() _batch_done(room) threading.Thread(target=_handle, daemon=True).start() def _batch_group_message(room: str, nickname: str, body: str) -> bool: """Add message to room batch. Returns True if batched, False if @mention (immediate).""" if f"@{_BOT_NICK}" in body or body.startswith(_BOT_NICK): return False formatted = f"[{nickname}]: {body}" with _batch_lock: if room in _batch_processing: _batch_pending.setdefault(room, []).append(formatted) return True timer = _batch_timers.pop(room, None) if timer: timer.cancel() _batch_entries.setdefault(room, []).append(formatted) t = threading.Timer(_BATCH_WINDOW, _fire_batch, args=[room]) t.daemon = True t.start() _batch_timers[room] = t return True # ═══════════════════════════════════════════════════════════════ # MAM Recovery — fetch recent history on startup # ═══════════════════════════════════════════════════════════════ _MAM_RECOVERY = True _MAM_RECOVERY_LOCK = threading.Lock() _MAM_MARK_DONE = False _MAM_TIMEOUT = 30 def _set_mam_done(): global _MAM_RECOVERY with _MAM_RECOVERY_LOCK: _MAM_RECOVERY = False def _is_mam_recovery() -> bool: if time.time() - _START_TIME > _MAM_TIMEOUT: global _MAM_RECOVERY with _MAM_RECOVERY_LOCK: if _MAM_RECOVERY: _MAM_RECOVERY = False log("(MAM recovery timed out, force-disabled)") return _MAM_RECOVERY with _MAM_RECOVERY_LOCK: return _MAM_RECOVERY # ═══════════════════════════════════════════════════════════════ # XML Escape # ═══════════════════════════════════════════════════════════════ def _escape(text: str) -> str: return (text.replace("&", "&").replace("<", "<") .replace(">", ">").replace('"', """)) # ═══════════════════════════════════════════════════════════════ # HTTP Bridge — health, presence, messages, POST send # ═══════════════════════════════════════════════════════════════ _HTTP_PORT = cfg["http_port"] _MSG_BUF: list[dict] = [] _MSG_BUF_LOCK = threading.Lock() _xmpp_ref = None # set after bot creation def _record_group_msg(nickname: str, body: str): ts = time.strftime("%H:%M:%S") with _MSG_BUF_LOCK: _MSG_BUF.append({"ts": ts, "from": nickname, "body": body}) if len(_MSG_BUF) > 200: _MSG_BUF[:] = _MSG_BUF[-150:] _BRIDGE_API_KEY = cfg.get('bridge_api_key', '') def _bridge_auth(self) -> bool: if not _BRIDGE_API_KEY: return True hdr = self.headers.get('X-Api-Key', '') if hdr == _BRIDGE_API_KEY: return True parsed = urllib.parse.urlparse(self.path) qs = urllib.parse.parse_qs(parsed.query) if qs.get('key', [None])[0] == _BRIDGE_API_KEY: return True return False class _BridgeHandler(http.server.BaseHTTPRequestHandler): def do_GET(self): parsed = urllib.parse.urlparse(self.path) # /health 免认证，供监控系统使用 if parsed.path == "/health": try: bot = _xmpp_ref session_ok = bot.session_started_event.is_set() if (bot and hasattr(bot, 'session_started_event')) else False socket_ok = bot.is_connected() if (bot and hasattr(bot, 'is_connected')) else False self._reply(200, { "ok": True, "xmpp_connected": session_ok or socket_ok, "agent": _agent_name, "jid": cfg["jid"], "uptime_sec": int(time.time() - _START_TIME), "muc_rooms": cfg["muc_rooms"], }) except Exception as e: self._reply(500, {"ok": False, "error": str(e)}) return if not _bridge_auth(self): self._reply(401, {'ok': False, 'error': 'unauthorized'}) return if parsed.path == "/muc": try: muc_info = {"rooms": {}} bot = _xmpp_ref if bot is not None and 'xep_0045' in bot.plugin: muc_plugin = bot.plugin['xep_0045'] for room_jid in cfg["muc_rooms"]: room_data = {"jid": room_jid, "participants": []} try: if room_jid in muc_plugin.rooms: room = muc_plugin.rooms[room_jid] for nick, info in room.get('roster', {}).items(): room_data["participants"].append({ "nick": nick, "jid": str(info.get('jid', '')), "affiliation": str(info.get('affiliation', '')), "role": str(info.get('role', '')), }) except Exception as e: room_data["error"] = str(e) muc_info["rooms"][room_jid] = room_data self._reply(200, muc_info) except Exception as e: self._reply(500, {"ok": False, "error": str(e)}) return if parsed.path.startswith("/presence"): jid_to_check = parsed.path[len("/presence/"):].strip() if not jid_to_check: self._reply(400, {"ok": False, "error": "missing JID"}) return try: info = {"jid": jid_to_check, "online": False, "resources": []} bot = _xmpp_ref if bot and hasattr(bot, 'client_roster'): roster = bot.client_roster if jid_to_check in roster: resources = list(roster[jid_to_check].resources.keys()) info["online"] = len(resources) > 0 info["resources"] = resources self._reply(200, info) except Exception as e: self._reply(500, {"ok": False, "error": str(e)}) return if parsed.path == "/messages": try: qs = urllib.parse.parse_qs(parsed.query) sender = qs.get("from", [None])[0] with _MSG_BUF_LOCK: msgs = list(_MSG_BUF) if sender: msgs = [m for m in msgs if m["from"] == sender] self._reply(200, {"ok": True, "count": len(msgs), "messages": msgs[-50:]}) except Exception as e: self._reply(500, {"ok": False, "error": str(e)}) return self._reply(404, {"ok": False, "error": "not found"}) def do_POST(self): if not _bridge_auth(self): self._reply(401, {'ok': False, 'error': 'unauthorized'}) return try: length = int(self.headers.get('Content-Length', 0)) body = json.loads(self.rfile.read(length)) path = urllib.parse.urlparse(self.path).path.rstrip('/') if path == '/rdp': action = body.get('action', '') if action == 'start': ok, msg = _rdp_enable() self._reply(200, {'ok': ok, 'message': msg}) elif action == 'stop': ok, msg = _rdp_disable() self._reply(200, {'ok': ok, 'message': msg}) elif action == 'status': self._reply(200, _rdp_status()) elif action == 'progress': self._reply(200, _rdp_progress()) elif action == 'enable_log': self._reply(200, _rdp_read_enable_log(body.get('lines', 80))) else: self._reply(400, {'ok': False, 'error': 'action must be start|stop|status|progress|enable_log'}) return # /usage endpoint — OpenCode Go usage monitor (read cache / trigger collection) # Pattern mirrors /rdp and /easytier. Two actions: # - status: return cached gateway/temp/usage_stats.json (instant, ~1ms) # - collect_now: spawn background thread running usage_collector.py (non-blocking, # refresh takes ~10-15s); client should poll status a few seconds later if path == "/usage": action = body.get('action', 'status') if action == 'status': self._reply(200, _usage_read_cache()) elif action == 'collect_now': _usage_trigger_async() self._reply(200, {'ok': True, 'message': 'collection triggered (takes ~10-15s); poll GET /usage action=status shortly', 'triggered_at': _now_iso()}) else: self._reply(400, {'ok': False, 'error': 'action must be status|collect_now'}) return # /easytier endpoint — execute EasyTier action locally (no XMPP DM) if path == "/easytier": action = body.get("action", "") if action == "start": _start_easytier() self._reply(200, {"ok": True, "message": "EasyTier started"}) elif action == "stop": _stop_easytier() self._reply(200, {"ok": True, "message": "EasyTier stopped"}) elif action == "status": running = _check_easytier() self._reply(200, {"ok": True, "running": running}) else: self._reply(400, {"ok": False, "error": "action must be start|stop|status"}) return # /send — 发送消息到 MUC 群聊 if path == "/send": to = body.get('to', '') if not to: rooms = cfg.get("muc_rooms", []) to = rooms[0] if rooms else "coregroup@conference.yoin.fun" msg = body.get('message', '') or body.get('body', '') msg_type = body.get('type', 'groupchat') if not msg: self._reply(400, {"ok": False, "error": "empty message"}) return bot = _xmpp_ref if bot: bot.send_message(mto=to, mbody=msg.strip(), mtype=msg_type) _record_group_msg(cfg["nick"], msg) log(f"[http] → [{to.split('@')[0]}]: {msg[:80]} (type={msg_type})") self._reply(200, {"ok": True}) return to = body.get('to', cfg["muc_rooms"][0]) msg = body.get('message', '') or body.get('body', '') msg_type = body.get('type', 'groupchat') if not msg: self._reply(400, {"ok": False, "error": "empty message"}) return safe = _escape(msg.strip()) bot = _xmpp_ref if bot: bot.send_message(mto=to, mbody=msg.strip(), mtype=msg_type) _record_group_msg(cfg["nick"], msg) log(f"[http] → [{to.split('@')[0]}]: {msg[:80]} (type={msg_type})") self._reply(200, {"ok": True}) except Exception as e: self._reply(500, {"ok": False, "error": str(e)}) def _reply(self, code, data): body = json.dumps(data, ensure_ascii=False).encode('utf-8') self.send_response(code) self.send_header('Content-Type', 'application/json; charset=utf-8') self.send_header('Content-Length', len(body)) self.end_headers() self.wfile.write(body) def log_message(self, format, *args): pass def _start_http_bridge(): _httpd = http.server.HTTPServer(('0.0.0.0', _HTTP_PORT), _BridgeHandler) _t = threading.Thread(target=_httpd.serve_forever, daemon=True) _t.start() log(f"HTTP bridge ready on :{_HTTP_PORT}") # ═══════════════════════════════════════════════════════════════ # Reply Processing (shared for all LLM responses) # ═══════════════════════════════════════════════════════════════ def _process_llm_reply(raw_reply: str, room: str): """Process LLM response: check silence/delay/exec/send.""" global _xmpp_ref if not raw_reply: _batch_done(room) return # Parse GRANT signal from LLM response _process_llm_grant(raw_reply) # ##delay:N## → schedule later delay_m = _DELAY_RE.search(raw_reply) if delay_m: sec = int(delay_m.group(1)) if delay_m.group(1) else _DELAY_DEFAULT _schedule_delayed(sec, room) _batch_done(room) return # ##exec:command## → run command, use output as reply exec_m = _EXEC_RE.search(raw_reply) if exec_m: output = _run_command(exec_m.group(1)) raw_reply = _EXEC_RE.sub(output, raw_reply, count=1) # Extract actual response reply_text = _extract_response(raw_reply) if reply_text: text = reply_text.strip() bot = _xmpp_ref if bot: bot.send_message(mto=room, mbody=text, mtype='groupchat') log(f"-> [{room.split('@')[0]}]: {text[:80]}") else: log(f"-> [{room.split('@')[0]}]: (silent)") _batch_done(room) # ═══════════════════════════════════════════════════════════════ # Group message handler # ═══════════════════════════════════════════════════════════════ def _handle_group_message(msg): """Process a groupchat message (runs in thread).""" global _COORDINATOR, _GRANTED, _REVOKED_UNTIL if _is_mam_recovery(): return msg_id = msg.get("id", "") if _is_duplicate(msg_id): return body = str(msg["body"]).strip() if not body: return full_from = str(msg["from"]) room = full_from.split("/")[0] nickname = full_from.split("/")[1] if "/" in full_from else "" # Self-message skip if nickname == cfg["nick"]: log(f"(self) {body[:80]}") return _record_group_msg(nickname, body) # Coordinator signals if _process_coordinator_signals(nickname, body): return # Revoke check is_revoked = time.time() < _REVOKED_UNTIL if is_revoked and _GRANTED == cfg["nick"]: _GRANTED = None is_revoked = False log(f"GRANT overrides REVOKE for {cfg['nick']}") if _check_shutup(body): return if is_revoked: body = f"【只读消息】你被收回发言权。只需了解内容。输出 __SILENT__。\n\n[核心群 {room}] {nickname} 说: {body}" # Batch or immediate (@mention) if _batch_group_message(room, nickname, body): log(f"[{room.split('@')[0]}] {nickname}: {body[:80]} (batched)") return log(f"[{room.split('@')[0]}] {nickname}: {body[:80]}") raw = _call_llm(body, full_from, is_group=True) _process_llm_reply(raw, room) # ═══════════════════════════════════════════════════════════════ # Private message handler # ═══════════════════════════════════════════════════════════════ def _handle_private_message(msg): """Process a private chat message.""" global _CALL_SEQ if msg["type"] == "groupchat": return msg_id = msg.get("id", "") if _is_duplicate(msg_id): return body = str(msg["body"]).strip() sender = str(msg["from"]).split("/")[0] log(f"<{sender}> {body[:80]}") if sender == cfg["jid"]: log("(skipped self)") return if time.time() < _REVOKED_UNTIL: log(f"(silenced) <{sender}> dropped") return if _check_shutup(body): return # ── Kanban routing ── _CALL_SEQ += 1 is_kanban = body.startswith('[Kanban]') target_sid = _KANBAN_SESSION_ID if is_kanban else cfg["session_id"] if is_kanban: log(f"📋 看板通知(#{_CALL_SEQ}): {body[:80]}") # ── EasyTier toggle ── if body.startswith('[EasyTier]'): action = body.replace('[EasyTier]', '').strip().lower() log(f"🔌 EasyTier command: {action}") if action == 'start': _start_easytier() reply_text = "[EasyTier] started on Windows" elif action == 'stop': _stop_easytier() reply_text = "[EasyTier] stopped on Windows" else: reply_text = f"[EasyTier] unknown action: {action}" bot = _xmpp_ref if bot: bot.send_message(mto=sender, mbody=reply_text, mtype='chat') log(f"-> {sender}: {reply_text}") return raw = _call_llm(body, sender, is_group=False, session_id=target_sid) if raw: reply = _extract_response(raw) if reply: text = reply.strip() bot = _xmpp_ref if bot: bot.send_message(mto=sender, mbody=text, mtype='chat') log(f"-> {sender}: {text[:80]}") # ═══════════════════════════════════════════════════════════════ # AgentBot Class # ═══════════════════════════════════════════════════════════════ import slixmpp class AgentBot(slixmpp.ClientXMPP): def __init__(self): super().__init__(cfg["jid"], cfg["password"]) # Connection settings self.enable_direct_tls = False self.enable_starttls = True self.auto_reconnect = True self.reconnect_max_delay = 10 self.whitespace_keepalive = True self.whitespace_keepalive_interval = 30 # SSL: accept self-signed certs ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE self.ssl_context = ctx # Event handlers self.add_event_handler("session_start", self._on_session_start) self.add_event_handler("message", self._on_any_message) self.add_event_handler("groupchat_message", self._on_group_msg) self.add_event_handler("disconnected", self._on_disconnected) self.add_event_handler("connected", self._on_connected) self.add_event_handler("session_end", self._on_session_end) self.add_event_handler("connection_failed", self._on_conn_failed) # MUC plugin self.register_plugin('xep_0045') def _on_connected(self, event): log("connection established") def _on_session_start(self, event): self.send_presence() self.get_roster() log(f"{cfg['jid']} online") # Register MAM plugin lazily try: self.register_plugin('xep_0313') except Exception: log("(MAM: xep_0313 not available)") # Join MUC rooms async def _join_all(): for room_jid in cfg["muc_rooms"]: try: self.plugin['xep_0045'].join_muc(room_jid, cfg["nick"]) presence = ( f"" f"" f"" f"" ) self.send_raw(presence) log(f"Joined {room_jid}") except Exception as e: log(f"MUC join failed {room_jid}: {e}") await asyncio.sleep(2) await asyncio.sleep(3) await self._fetch_mam_history() asyncio.ensure_future(_join_all()) async def _fetch_mam_history(self): """Query MAM for recent MUC messages to rebuild context.""" if 'xep_0313' not in self.plugin: log("(MAM: no plugin)") _set_mam_done() return # MAM recovery used in _on_session_start try: for room_jid in cfg["muc_rooms"]: log(f"(MAM: querying {room_jid} for last 50 messages...)") results = await self.plugin['xep_0313'].retrieve( jid=room_jid, rsm={'max': 50}, ) count = 0 for msg in results['mam']['results']: forwarded = msg['mam_result']['forwarded'] body = str(forwarded['stanza']['body'] or '').strip() if not body: continue nick = str(forwarded['stanza']['from']).split('/')[-1] if '/' in str(forwarded['stanza']['from']) else '?' # Feed into context for chat_bridge (xxm) if _IS_CHAT_BRIDGE: role = 'user' if nick != cfg["nick"] else 'assistant' try: _bridge._append_to_log(role, f"[{nick}]: {body[:300]}") except Exception: pass count += 1 log(f"(MAM: loaded {count} msgs from {room_jid})") _set_mam_done() log("(MAM recovery complete)") except Exception as e: log(f"(MAM error: {e})") _set_mam_done() def _on_group_msg(self, msg): threading.Thread(target=_handle_group_message, args=[msg], daemon=True).start() def _on_any_message(self, msg): threading.Thread(target=_handle_private_message, args=[msg], daemon=True).start() def _on_session_end(self, event): log(f"session ended") def _on_conn_failed(self, event): log(f"connection failed: {event}") def _on_disconnected(self, event): log(f"disconnected, reconnecting...") # ═══════════════════════════════════════════════════════════════ # Main # ═══════════════════════════════════════════════════════════════ def main(): log(f"Starting {cfg['jid']} ({cfg['name_cn']}) — agent={_agent_name}") if _IS_CHAT_BRIDGE: log(f" LLM: chat_bridge (session={cfg['session_id']})") else: log(f" LLM: Hermes API ({cfg['gateway']})") log(f" Server: {cfg['server']}:{cfg['port']}") log(f" Rooms: {cfg['muc_rooms']}") bot = AgentBot() global _xmpp_ref _xmpp_ref = bot _start_http_bridge() _start_usage_auto_timer() bot.connect(host=cfg["server"], port=cfg["port"]) log(f"Connecting {cfg['jid']}@{cfg['server']}:{cfg['port']}") loop = asyncio.get_event_loop() async def _status_check(): skipped = 0 while True: await asyncio.sleep(60) alive = bot and bot.is_connected() if hasattr(bot, 'is_connected') else False if not alive: skipped += 1 log(f"[R02] XMPP 连接已断开 ({skipped}/3)，等待重连...") if skipped >= 3: log("[R02] 连续 3 次检测到连接断开，退出进程让 systemd 重启") os._exit(1) else: if skipped > 0: log(f"(连接已恢复，跳过 {skipped} 次检测)") skipped = 0 asyncio.ensure_future(_status_check()) try: loop.run_forever() except KeyboardInterrupt: log("Shutdown by user") except Exception as e: log(f"!!! MAIN LOOP CRASH: {e}") import traceback log(f"!!! {traceback.format_exc()[:500]}") raise if __name__ == "__main__": main()