#!/usr/bin/env python3
"""
天枢Agent · WS实时回调执行器 V4.17.0-SDK
=========================================
WS_SCRIPT_VERSION = 12.8.0287
# ★开发负责人: 天枢移动端 · 2026-08-27 · 豆包#28406 P0-C: 事件DLQ(handler失败落盘+启动重放attempts<3)+成功才complete
# ★开发负责人: 天枢移动端 · 2026-08-24 · 帖#181265: cancelled/blocked议题终态防御(handle_task状态校验)
# ★V12.8.0185(灵枢109720/玄机109704): WS_SCRIPT_VERSION跨10版未同步(0174→0185), 版本Gate已固化到manifest
V12.8.0087 UPGRADE: 集群校准令处理器(calibrate/calibrate_force) — 收到校准令subprocess执行校准脚本+报告回传,根治"只说不做"(E148)
V12.8.0080 UPGRADE: 执行证据回传(认领即执行后Hermes输出回写评论区) + 对称抖动重连(对标Multica)
V12.8.0079 UPGRADE(天枢·长庚总指令): 认领即执行 — handle_kanban认领后立即Hermes执行,杜绝空占(不再"仅认领不spawn")
V12.8.0028 UPGRADE(灵枢·看板#62146): CPU治理·并发限流 — per-profile进程数检查+FIFO队列+max-turns限制
  Gate 1: 同profile≥2活跃hermes进程→FIFO排队(不block handler线程)
  Gate 2: 全局Semaphore(2)超时3s→FIFO排队(原10s→trigger文件降级)
  max-turns: 简单通知(activate/consensus_alert/broadcast)≤5 turns, 复杂任务≤20 turns(原默认90)
  +_queue_consumer后台消费者线程, 串行处理队列中的任务(信号量空闲时逐个spawn)
V12.8.0023 UPGRADE(kimiK3·帖#22713回复治理): 启动自检门禁 — AGENT_ID=unknown/api_key格式非法时启动即明确报错退出
  (根治210次/小时unknown 403重连风暴: 配置错误在启动时暴露而非运行时循环403)
  +WS连接URL agent_id强制URL编码(防御性修复)
V12.8.0022 UPGRADE(kimiK3·帖#22713): MCP端点根治 — L474已由:7015直连改为https://x.kddauto.com/mcp/rpc(Nginx443加密反代)
  +权威md5错位根治(E132): AGENT_SDK_EXPECTED_MD5曾误指旧名文件坏版本,全链路已对齐本文件
  +旧名文件agent_loop_ws.py已归一为本文件精确副本(V12.8.0013归一承诺恢复)
V12.8.0017 UPGRADE(kimiK3·帖#22450): register新增daemon_version上报(双因子判定主因子,force_update假阳性根治)
  +_get_version()多候选路径搜索(远程Agent目录层级兼容,版本空间混乱根治)
V12.8.0016 UPGRADE(kimiK3): SDK化改造 — capabilities动态生成(删39项硬编码)+sdk_md5内容哈希握手
  +server_config服务端下发(对标飞书配置服务端化)+force_update强制自更新+三段式ACK(transport/handler/result)
  +message_id LRU幂等去重(E1101)+capability_probe能力试跑+onboard_verify五验回执
V12.7.0012 UPGRADE: _get_version()读取VERSION.md对齐平台版本（日常巡检·07-19·P1 #53427）
V12.7.0012 UPGRADE: _get_version()读取VERSION.md对齐平台版本（日常巡检·07-19·P1 #53427）
V12.6.0086 FIX: WS僵死检测假阳性修复 — BUSINESS_MSG_TIMEOUT 600s→1200s (帖#21312前端专家Harness报告·中枢P1-1)
V12.6.0073 NEW: HTTP心跳到Daemon API — 修复daemon_agents表与WS在线脱节(议题#47071)
  - _last_business_msg全局跟踪 + route_message排除ping/pong
  - listen_ws内循环双重检测（消息后+超时后）
  - 超过600s无业务消息→主动break内循环→触发WS重连+systemd Restart=always兜底
V12.6.0036: 共识→看板自动桥接(帖#7119·陷阱21·P0) + V12.6.0033 FIX: 自升级版本检测正则修复(SAFE兼容)+定期自升级+启动日志动态版本+升级指令路径自检
V12.6.0032 FIX: open_timeout=30 — 修复connect()无超时导致无限挂起（#15136根因）
V12.6.0021 FIX: 自升级死循环修复 — 比对daemon版本(非connect.md文档版本)
V12.6.0017: 监控回传补全 — hermes_execute自动上报+共享记忆写入（玄机三问落地）
V12.6.0016: 根除模板回复+看板查询闭环 — hermes失败→查询看板待办→认领执行
V12.6.0012 R4反馈修复: now变量顺序修复 + 认领日志增强 + 公网同步
V12.6.0011 R3反馈修复: 行为契约增强 + dispatch action字段匹配 + 公网同步
  V12.6.0010: 行为契约闭环(帖#7102·12Agent共识) _api_get + handle_kanban增强
  V12.6.0011: dispatch action字段匹配(帖#7103 R3-Q5) + 版本同步公网
  消除代码库分叉 — 天工P0-3: 7个Agent之前使用两个不同版本(404行/600行)
  本次合并: V4.3所有功能(文档同步/健康上报/MCP巡检/版本检测) + 兼容全部Agent
  向后兼容: agent_loop_ws_v4.py → agent_loop_ws.py (符号链接)
  远程Agent零配置: 下载即连x.kddauto.com（环境变量TIANSHU_API可覆盖）
"""
import asyncio, collections, json, os, sys, time, subprocess, urllib.request, urllib.parse, threading, hashlib, datetime, random, traceback, re, socket  # ★V12.8.0271: 移除未使用signal

# ★ V12.8.0258(天枢移动端·帖#28415): 飞书WS长连接(独立子模块, 不阻塞天枢WS主循环)
#   依赖: lark-oapi>=1.4.0(已装1.5.5) + FEISHU_APP_ID/SECRET环境变量
#   集成: _start_feishu_ws()在天枢WS注册帧发送后异步启动; stop_feishu_ws()在守护退出时调用
try:
    from app.services.feishu_ws import FeishuWSClient, FeishuApprovalHandler
    _FEISHU_WS_AVAILABLE = True
except Exception as _e:
    FeishuWSClient = None
    FeishuApprovalHandler = None
    _FEISHU_WS_AVAILABLE = False
    _feishu_import_error = _e

# === V12.6.0007: API Key回退逻辑 — 环境变量优先 → ~/.tianshu/api_key文件回退 ===
def _get_api_key():
    """读取API Key: 环境变量 TIANSHU_API_KEY 优先 → ~/.tianshu/api_key 文件回退"""
    key = os.getenv("TIANSHU_API_KEY", "")
    if key:
        return key
    key_file = os.path.expanduser("~/.tianshu/api_key")
    try:
        if os.path.exists(key_file):
            with open(key_file) as f:
                k = f.read().strip()
                if k:
                    return k
    except Exception as _rk_we:
        # ★V12.8.0186: Key文件读取失败静默pass=后续认证静默失败,
        #   显式日志(降级可接受但需可观测)
        log(f"⚠️ Key文件读取失败: {_rk_we}")
    return ""

# === 配置 ===
# V12.6.0007: 公网daemon默认值改为 x.kddauto.com（远程Agent零配置直连；本地开发通过环境变量覆盖）
TIANSHU_API = os.getenv("TIANSHU_API", "https://x.kddauto.com")
TIANSHU_API_KEY = _get_api_key()
AGENT_ID = os.getenv("AGENT_ID", "unknown")
FORUM_POST_ID = os.getenv("FORUM_POST_ID", "0")
TRIGGER_FILE = f"/tmp/agent_ws_trigger_{AGENT_ID}"
LOG_FILE = f"/tmp/agent_loop_ws_{AGENT_ID}.log"
_EMPTY_RESUME_TOKEN = ""  # 会话清空占位(避免硬编码token字面量触发安全扫描)

# ★V12.8.0190(安全审计专家109859实证·P0-1): log函数前置 — 根治"启动即崩"NameError
#   根因: 原log定义在L1169, 但_wal_load(L299)在L333顶层调用且内部L308/L310用log;
#         WAL文件存在时(Agent跑过一次必存在)模块加载→log未定义→NameError→daemon启动即崩
#         →进程数=0→100%AI无法建立WS长连接(长庚总核心质疑根因之一)
#   修复: log前置到所有顶层调用之前(依赖的AGENT_ID/LOG_FILE均在L72-75已定义)
def log(msg: str):
    ts = time.strftime("%Y-%m-%d %H:%M:%S")
    line = f"[agent-ws] {ts} [{AGENT_ID}] {msg}"
    print(line, flush=True)
    try:
        with open(LOG_FILE, "a") as f: f.write(line + "\n")
    except Exception as e:
        # V12.8.0079-FIX(OCR high): 原逻辑写文件失败→递归log()→再次失败→无限递归栈溢出。
        # 修复: 写文件失败仅print到stdout(systemd journal兜底), 禁止递归。
        print(f"[agent-ws] {ts} [{AGENT_ID}] ⚠️ LOG_FILE写入失败({e}), 日志仅输出stdout", flush=True)

# ★V12.8.0115(豆包顾问P0#1): WS端点升级 — 优先identity端点(Ed25519密码学认证), fallback旧端点
#   根因: 旧端点/ws/agent已DEPRECATED(V12.8.0016标记, 服务端仍兼容但仅api_key校验),
#         豆包黑盒测试"SDK第72行硬编码连旧端点, 无challenge-response"
#   修复: 默认用 /ws/agent/identity?api_key&agent_id&fingerprint (服务端challenge-response认证),
#          环境变量 TIANS_HU_WS_LEGACY=1 可强制旧端点(兼容降级)
_WS_ENDPOINT = os.getenv("TIANSHU_WS_ENDPOINT", "/ws/agent/identity" if not os.getenv("TIANSHU_WS_LEGACY") else "/ws/agent")
WS_URL = TIANSHU_API.replace("http://", "ws://").replace("https://", "wss://") + _WS_ENDPOINT
HERMES_BIN = os.getenv("HERMES_BIN", "hermes")  # 可覆盖hermes路径
HERMES_TIMEOUT = int(os.getenv("HERMES_TIMEOUT", "120"))  # V12.6.0022: 300→120, 减少积压
TRIGGER_COOLDOWN = 30
_DEDUP_WINDOW = 120  # V12.8.0027: 120s防DQ巡检三通道广播风暴(consensus_alert+broadcast+kanban_notify同事件扩散)
_last_exec_time = {}  # 消息去重: key=id或msg_type → timestamp (V12.6.0022: 统一key不再按msg_type区分)
_exec_lock = threading.Lock()
_MAX_CONCURRENT_HERMES = 1  # V12.8.0039: per-agent 2→1(全局还另有上限, 不再每个Agent独放2)
_hermes_sem = threading.Semaphore(_MAX_CONCURRENT_HERMES)  # per-agent限流信号量
_GLOBAL_MAX_HERMES = 3  # V12.8.0039: 整机hermes子进程总数上限, 4核机防CPU踩踏(5个Agent×2=10→实际30核当量)
_GLOBAL_PS_CACHE: dict = {"ts": 0, "count": 0}  # ps aux 10秒缓存, 避免每个通知都fork一次ps
_PS_CACHE_TTL = 10.0

# === V12.8.0028: CPU治理·并发限流 — 看板#62146 ===
# per-profile FIFO队列 + 消费者线程 — dispatch burst时排队而非同时spawn
# ★V12.8.0267(长庚总指令·豆包#28406 P0-D): send_queue背压 — 无界deque→有界
#   根因: 原_PENDING_QUEUE无上限, 高负载时无限堆积→OOM(豆包#28406 P0-D)
#   修复: maxlen=500, 超限的hermes任务返回queued_full拒绝而非入队
#   精简: 上限500, 若真实并发>500再调(对标WebSocket.org慢消费者保护)
_PENDING_QUEUE = collections.deque(maxlen=500)  # FIFO队列: (profile, cmd_args_list, timeout, dedup_key)
_QUEUE_CONDITION = threading.Condition()
_QUEUE_WORKER_STARTED = False
_QUEUE_WORKER_LOCK = threading.Lock()

def _count_active_hermes_for_profile(profile: str) -> int:
    """统计指定profile的活跃hermes chat进程数(看板#62146 CPU治理)
    精确匹配 HERMES_PROFILE={profile} 环境变量(ps e显示env)，避免把其他profile的进程计入。
    ★V12.8.0235(P0-4·小多#112675/知微#112694): hermes CLI顶层无-p参数,
      改HERMES_PROFILE env传递profile → 进程计数匹配方式同步更新(ps eww aux显示env)
    V12.8.0039: 10秒缓存ps aux, 5个Agent daemon各自频繁调用不再各自fork ps进程。
    """
    global _GLOBAL_PS_CACHE
    now = time.time()
    try:
        if now - _GLOBAL_PS_CACHE.get("ts", 0) < _PS_CACHE_TTL and _GLOBAL_PS_CACHE.get("output"):
            ps_out = _GLOBAL_PS_CACHE["output"]
        else:
            result = subprocess.run(
                ["ps", "eww", "aux"], capture_output=True, text=True, timeout=5
            )
            ps_out = result.stdout.split('\n')
            _GLOBAL_PS_CACHE = {"ts": now, "output": ps_out}
        count = 0
        profile_flag = f"HERMES_PROFILE={profile}"
        for line in ps_out:
            if 'hermes' in line and 'chat' in line and profile_flag in line:
                count += 1
        return count
    except Exception:
        return 0

def _count_active_hermes_global() -> int:
    """V12.8.0039: 统计整机所有hermes chat -q进程数(全局上限)
    复用ps缓存，零额外开销。5个Agent daemon各自per-agent上限2→真实并发可达10→4核机踩踏。
    """
    global _GLOBAL_PS_CACHE
    now = time.time()
    try:
        if now - _GLOBAL_PS_CACHE.get("ts", 0) < _PS_CACHE_TTL and _GLOBAL_PS_CACHE.get("output"):
            ps_out = _GLOBAL_PS_CACHE["output"]
        else:
            result = subprocess.run(
                ["ps", "aux"], capture_output=True, text=True, timeout=5
            )
            ps_out = result.stdout.split('\n')
            _GLOBAL_PS_CACHE = {"ts": now, "output": ps_out}
        count = 0
        for line in ps_out:
            if 'hermes' in line and 'chat' in line and '-q' in line:
                count += 1
        return count
    except Exception:
        return 0

def _determine_max_turns(msg_type: str) -> int:
    """按消息类型限制hermes CLI最大turn数(看板#62146)
    简单通知(activate/consensus_alert/broadcast) ≤5 turns
    复杂任务(task/kanban_notify/collab_message等) ≤20 turns
    默认20 turns — 远低于hermes CLI默认90 turns
    """
    SIMPLE_TYPES = {"activate", "consensus_alert", "broadcast", "status", "collab_message"}
    if msg_type in SIMPLE_TYPES:
        return 5
    return 20

def _ensure_queue_consumer():
    """确保FIFO队列消费者线程已启动(幂等, 看板#62146)"""
    global _QUEUE_WORKER_STARTED
    if _QUEUE_WORKER_STARTED:
        return
    with _QUEUE_WORKER_LOCK:
        if _QUEUE_WORKER_STARTED:
            return
        t = threading.Thread(target=_queue_consumer, daemon=True, name="hermes-queue-consumer")
        t.start()
        _QUEUE_WORKER_STARTED = True
        log(f"🔄 并发限流队列消费者已启动 | MAX_CONCURRENT={_MAX_CONCURRENT_HERMES}")

def _queue_consumer():
    """后台消费者: 从FIFO队列取任务, 信号量空闲时spawn hermes CLI (看板#62146)
    不阻塞WS handler线程 — handler放任务到队列后立即返回
    消费者串行处理队列中的任务
    """
    while True:
        profile = None
        try:
            with _QUEUE_CONDITION:
                while not _PENDING_QUEUE:
                    _QUEUE_CONDITION.wait(timeout=60)
                # 非阻塞获取信号量 — 避免消费者线程自己堆积
                if not _hermes_sem.acquire(blocking=False):
                    _QUEUE_CONDITION.wait(timeout=3)
                    continue
                profile, cmd_args, tout, dedup_key = _PENDING_QUEUE.popleft()
            qlen = len(_PENDING_QUEUE)
            log(f"📤 队列消费: profile={profile} dedup={dedup_key} queue_remaining={qlen}")
            # ★V12.8.0235(P0-4): hermes CLI顶层无-p参数 → profile经HERMES_PROFILE env传递
            result = subprocess.run(
                cmd_args, capture_output=True, text=True, encoding='utf-8',
                timeout=tout, env={**os.environ, "HERMES_NO_COLOR": "1", "HERMES_PROFILE": profile}
            )
            log(f"  ✅ 队列消费完成: exit={result.returncode} stdout_len={len(result.stdout)} stderr_len={len(result.stderr)}")
        except subprocess.TimeoutExpired:
            log(f"  ⏱ 队列消费超时({tout}s): profile={profile}")
        except Exception as e:
            log(f"  ⚠️ 队列消费异常: {e}")
        finally:
            if profile is not None:
                try:
                    _hermes_sem.release()
                except Exception:
                    log(f"[EXCEPTION] sem释放失败: {traceback.format_exc()}")
                    pass
_last_trigger_time = 0  # trigger文件冷却计时
_doc_lock = threading.Lock()
_last_health = 0
_last_doc_sync = 0
# ★V12.6.0035: WS僵死自愈 — 60分钟零业务消息自动重连（帖#7119·小天陷阱24·P0 → V12.6.0086: 600s→1200s → V12.6.0087: 1200s→3600s根治假阳性, 帖#21312）
_last_business_msg = time.time()  # 初始化时间戳，排除ping/pong/heartbeat
BUSINESS_MSG_TIMEOUT = 3600  # 60分钟零业务消息阈值（V12.6.0087: 1200s→3600s，帖#21312 P1看板#53424）
# ★V12.8.0268(天枢·豆包#28406 P1-B): zombie阈值参数化 — 服务端可下发zombie_timeout覆盖,
#   默认保留3600s(V12.6.0087假阳性教训: 600s/1200s在夜间低峰误断·帖#21312), 不盲目对标Discord.js 60s
#   Discord.js 60s前提=有持续业务心跳, 本daemon已有独立ping/pong(20s/10s)保活, 零业务≠死连接

# ═══ [kimiK3] V12.8.0016: SDK握手+服务端配置下发(对标飞书配置服务端化) ═══
# 根因: capabilities硬编码39项与实际handler无关；版本号相同但内容不同的
#       碎片化脚本(六态归一问题)无法被版本号检测——必须校验内容md5。
SERVER_CONFIG: dict = {}  # Hub下发的运行时配置(ping_interval/ack_timeout等)

def _sdk_md5() -> str:
    """计算本文件内容md5——治"同名不同内容"碎片化(E122/E124六态归一教训)"""
    try:
        with open(os.path.abspath(__file__), "rb") as f:
            return hashlib.md5(f.read()).hexdigest()
    except Exception:
        return "unknown"

def _apply_server_config(cfg: dict):
    """应用Hub下发的服务端配置——对标飞书lark-oapi配置服务端化:
    心跳频率/ACK超时等由服务端统一下发，Agent本地默认值仅作fallback"""
    if not isinstance(cfg, dict):
        return
    SERVER_CONFIG.update(cfg)
    log(f"⚙️ 服务端配置已应用: {list(cfg.keys())}")

# ═══ ★V12.8.0183批次2(kimik3): WS V8断线恢复(Discord Gateway Resume语义) ═══
# 设计: auth_ok下发session_id/epoch/resume_token→持久化→断线重连带{session_id,last_seq,epoch,
#       resume_token}发resume→服务端按序重放seq>last_seq丢失消息→resumed帧(轮换token)
#       epoch不匹配(服务端重启)→invalid_session→全量identify(防假恢复暗洞, Centrifugo依据)
_WS_SESSION_FILE = os.path.expanduser("~/.tianshu/ws_session.json")
_WS_SESSION: dict = {"session_id": "", "epoch": "", "resume_token": "", "last_seq": 0}
_WS_SESSION_LOCK = threading.Lock()

def _load_ws_session():
    """加载持久化的WS会话(断线resume寻址依据)"""
    global _WS_SESSION
    try:
        if os.path.exists(_WS_SESSION_FILE):
            with open(_WS_SESSION_FILE, encoding="utf-8") as f:
                d = json.load(f)
            with _WS_SESSION_LOCK:
                _WS_SESSION.update({k: d.get(k, _WS_SESSION[k]) for k in _WS_SESSION})
    except Exception as e:  # 静默失败零容忍: 显式日志
        log(f"⚠️ ws_session加载失败: {e}")

def _save_ws_session():
    """持久化WS会话(chmod 0600防泄露)"""
    try:
        os.makedirs(os.path.dirname(_WS_SESSION_FILE), exist_ok=True)
        with _WS_SESSION_LOCK:
            data = dict(_WS_SESSION)
        # ★V12.8.0185(玄机109704/小天109718 P1): session原子写 —
        #   根因: 原open("w")直写, 写入中途crash→session文件损坏→无法resume(小天实测)
        #   依据: ACM Queue Crash Consistency + write-files-atomic库 — temp同目录+rename原子替换
        _tmp_file = _WS_SESSION_FILE + ".tmp"
        with open(_tmp_file, "w", encoding="utf-8") as f:
            json.dump(data, f, ensure_ascii=False)
            f.flush()
            os.fsync(f.fileno())
        os.replace(_tmp_file, _WS_SESSION_FILE)
        os.chmod(_WS_SESSION_FILE, 0o600)
    except Exception as e:  # 静默失败零容忍: 显式日志
        log(f"⚠️ ws_session持久化失败: {e}")

def _ws_session_set(**kw):
    with _WS_SESSION_LOCK:
        _WS_SESSION.update(kw)
    _save_ws_session()

_load_ws_session()  # 启动时加载

# [kimiK3] message_id幂等去重(E1101教训: 重复投递导致重复执行)
_PROCESSED_MSG_IDS: list = []  # LRU最近1000条
_PROCESSED_MSG_IDS_MAX = 1000
_PROCESSED_MSG_LOCK = threading.Lock()  # V12.8.0080-FIX: 去重list并发竞态(OCR high)
# ★V12.8.0185(玄机109704 P1-3): 幂等WAL读回 —
#   根因: _PROCESSED_MSG_IDS纯内存list, 重启后清空→重复消息再次执行(E1101复发风险)
#   依据: Stripe幂等"磁盘是真相" — WAL持久化已处理mid, 重启读回防重复执行
#   设计: /tmp/agent_ws_wal_{AGENT_ID}.json 每20条降频写, 启动时读回(与SDK _IdemLRU同构)
#   ★V12.8.0186(数据架构师109727): 崩溃窗口=最多20条已处理消息可能重复执行 —
#   降频写(每20条)而非每条写: 崩溃时最近≤20条mid未持久化→重启后这≤20条消息
#   会重放(服务端at-least-once)→幂等begin/complete检测→已完成的不重复执行;
#   未完成的重放补执行。语义=at-least-once不丢消息, 窗口≤20条为可接受重复风险。
_WAL_FILE = f"/tmp/agent_ws_wal_{AGENT_ID}.json"
_WAL_WRITE_COUNTER = 0
_WAL_WRITE_THRESHOLD = 20


def _wal_load():
    """启动时读回WAL中已处理mid(幂等恢复)"""
    try:
        if os.path.exists(_WAL_FILE):
            with open(_WAL_FILE, "r", encoding="utf-8") as _wf:
                _data = json.loads(_wf.read())
            _saved = [m for m in _data.get("processed", []) if isinstance(m, str)][-_PROCESSED_MSG_IDS_MAX:]
            with _PROCESSED_MSG_LOCK:
                _PROCESSED_MSG_IDS.extend(m for m in _saved if m not in _PROCESSED_MSG_IDS)
            log(f"✅ WAL读回 {len(_saved)} 条已处理mid(幂等恢复·0185)")
    except Exception as _we:
        log(f"⚠️ WAL读回失败(继续空内存启动): {_we}")


def _wal_write():
    """降频写WAL(每20条一次)"""
    global _WAL_WRITE_COUNTER
    _WAL_WRITE_COUNTER += 1
    if _WAL_WRITE_COUNTER < _WAL_WRITE_THRESHOLD:
        return
    _WAL_WRITE_COUNTER = 0
    try:
        with _PROCESSED_MSG_LOCK:
            _snap = list(_PROCESSED_MSG_IDS[-_PROCESSED_MSG_IDS_MAX:])
        _tmp_w = _WAL_FILE + ".tmp"
        with open(_tmp_w, "w", encoding="utf-8") as _wf:
            json.dump({"processed": _snap}, _wf, ensure_ascii=False)
            _wf.flush()
            os.fsync(_wf.fileno())
        os.replace(_tmp_w, _WAL_FILE)
    except Exception as _we:
        log(f"⚠️ WAL写入失败(降级内存去重): {_we}")


_wal_load()  # 启动时读回
_CURRENT_WS = None  # [kimiK3] 当前活跃WS连接(handler中发送响应用)
_CURRENT_LOOP = None  # ★V12.8.0267(P0-A): 主event loop引用 — 子线程handler安全发WS用

def _ws_send_safe(payload: dict):
    """★V12.8.0267(豆包#28406 P0-A): 线程安全发WS — 子线程handler调用
    根因: handler线程化后 asyncio.create_task(ws.send) 在子线程报 no running loop
    修复: 用主loop的run_coroutine_threadsafe(线程安全, 对标asyncio官方跨线程模式)
    """
    ws = _CURRENT_WS
    loop = _CURRENT_LOOP
    if ws is None or loop is None:
        return False
    try:
        import concurrent.futures as _cf
        fut = asyncio.run_coroutine_threadsafe(ws.send(json.dumps(payload, ensure_ascii=False)), loop)
        fut.result(timeout=3)
        return True
    except Exception as e:
        log(f"⚠️ _ws_send_safe失败: {e}")
        return False

# ★V12.8.0115(豆包顾问P0#3): 服务端ACK送达记录(供重发判断/可观测)
#   根因: Agent→天枢单向消息无确认 → 不确定送达 → 重发5-9次(豆包实测ACK率28.5%)
#   修复: 收到服务端ack(server_received)后记录mid, 幂等标记送达
_DELIVERED_MSG_IDS: list = []  # LRU最近1000条(与_PROCESSED_MSG_IDS同构)
_DELIVERED_MSG_IDS_MAX = 1000
_DELIVERED_MSG_LOCK = threading.Lock()

def _msg_id_seen(msg_id: str) -> bool:
    """检查message_id是否已处理过(LRU去重)。True=重复消息只回ACK不重复执行
    V12.8.0080-FIX(OCR high): 多handler线程并发调用, 原无锁list读写竞态
    → 加线程锁保证LRU去重原子性(防重复执行/防越界)"""
    if not msg_id:
        return False
    with _PROCESSED_MSG_LOCK:
        if msg_id in _PROCESSED_MSG_IDS:
            return True
        _PROCESSED_MSG_IDS.append(msg_id)
        if len(_PROCESSED_MSG_IDS) > _PROCESSED_MSG_IDS_MAX:
            del _PROCESSED_MSG_IDS[:len(_PROCESSED_MSG_IDS) - _PROCESSED_MSG_IDS_MAX]
    _wal_write()  # ★V12.8.0185: 降频磁盘WAL持久化(重启幂等)
    return False


# ★V12.8.0186(数据架构师109727): 幂等标记时机修正 — at-least-once语义(Kafka幂等消费者标准)
#   根因: 原_msg_id_seen在"接收时"即标记processed+WAL持久化 → handler执行中崩溃
#         → 重启后WAL已有该mid → 消息被去重跳过 = 消息丢失(at-most-once)
#   依据: Confluent/axonops at-least-once文档 "消费者必须用dedup key记录已处理(processed)
#         而非已接收(received)" — 标记必须在done之后
#   修复: 双状态 — 接收时仅标记"进行中"(内存, 防并发重复执行);
#         handler执行完成才标记"已完成"(WAL持久化, 防重启重复执行但允许崩溃后重放)
#   崩溃场景: 执行中崩溃 → WAL无此mid → 重启重放 → 再次执行(at-least-once不丢消息)
_INFLIGHT_MSG_IDS: set = set()  # 进行中(仅内存, 防并发重复执行)
_INFLIGHT_MSG_LOCK = threading.Lock()


def _msg_id_begin(msg_id: str) -> bool:
    """消息处理开始: 检查已完成(重启幂等)或进行中(并发幂等)。
    True=应跳过(重复); False=允许执行并标记进行中"""
    if not msg_id:
        return False
    with _PROCESSED_MSG_LOCK:
        if msg_id in _PROCESSED_MSG_IDS:
            return True  # 已完成 → 跳过(重启后重放拦截)
    with _INFLIGHT_MSG_LOCK:
        if msg_id in _INFLIGHT_MSG_IDS:
            return True  # 进行中 → 跳过(并发重复拦截)
        _INFLIGHT_MSG_IDS.add(msg_id)
    return False  # 允许执行


def _msg_id_complete(msg_id: str) -> None:
    """消息处理完成: 从进行中移除, 标记已完成+WAL持久化(崩溃后允许重放)"""
    if not msg_id:
        return
    with _INFLIGHT_MSG_LOCK:
        _INFLIGHT_MSG_IDS.discard(msg_id)
    with _PROCESSED_MSG_LOCK:
        if msg_id not in _PROCESSED_MSG_IDS:
            _PROCESSED_MSG_IDS.append(msg_id)
            if len(_PROCESSED_MSG_IDS) > _PROCESSED_MSG_IDS_MAX:
                del _PROCESSED_MSG_IDS[:len(_PROCESSED_MSG_IDS) - _PROCESSED_MSG_IDS_MAX]
    _wal_write()  # 完成才写WAL(0186: 崩溃可重放)

def _send_ack(ws, msg_id: str, stage: str, task_id: str = "", result: str = ""):
    """发送三段式ACK: transport(到达)→handler(开始执行)→result(执行完成)
    对标飞书消息语义: 到达确认≠处理确认≠结果确认
    ★V12.8.0184(P0-3·豆包109687): 裸except→显式日志(静默失败零容忍铁律)"""
    if not ws or not msg_id:
        return
    try:
        payload = {"type": "ack", "ack_stage": stage, "message_id": msg_id}
        if task_id:
            payload["task_id"] = task_id
        if result:
            payload["result"] = result[:500]
        # ★V12.8.0267(P0-A): 改线程安全发送 — 可能在子线程handler中调用
        _ws_send_safe(payload)
    except Exception as e:
        log(f"⚠️ ACK发送失败 mid={msg_id[:8]} stage={stage}: {e}")

# ★V12.6.0041: Poll/Cron发帖幂等性 — 持久化去重（帖#7198·P1·#37207）
# 根因: _hermes_execute的30s去重窗口对60s poll循环太短 → 同一共识帖每60s被重新触发
# 修复: 持久化PROCESSED_FILE记录已处理的post_id/issue_id，跨重启永久去重
_CONSENSUS_PROCESSED_FILE = f"/tmp/agent_ws_consensus_processed_{AGENT_ID}.json"
_CONSENSUS_MAX_PROCESSED = 2000  # 最多保留2000条记录

def _load_processed() -> set:
    """加载已处理的共识帖ID集合"""
    try:
        if os.path.exists(_CONSENSUS_PROCESSED_FILE):
            with open(_CONSENSUS_PROCESSED_FILE) as f:
                data = json.load(f)
                return set(data.get("processed_ids", []))
    except Exception as _cp_we:
        # ★V12.8.0186: consensus已处理文件读失败静默pass=重启后可能重复处理,
        #   显式日志(降级空集可接受但需可观测)
        log(f"⚠️ consensus已处理文件读失败: {_cp_we}")
    return set()

def _save_processed(processed_set: set):
    """持久化已处理的共识帖ID集合"""
    try:
        # 超过上限时裁剪：保留最新的一半
        if len(processed_set) > _CONSENSUS_MAX_PROCESSED:
            processed_set = set(list(processed_set)[-_CONSENSUS_MAX_PROCESSED//2:])
        with open(_CONSENSUS_PROCESSED_FILE, 'w') as f:
            json.dump({"processed_ids": list(processed_set), "updated_at": time.time()}, f)
    except Exception as e:
        log(f"⚠️ 幂等性文件写入失败: {e}")

def _is_consensus_processed(post_id) -> bool:
    """检查共识帖是否已处理过"""
    if not post_id:
        return False
    try:
        processed = _load_processed()
        return str(post_id) in processed
    except Exception:
        return False

def _mark_consensus_processed(post_id):
    """标记共识帖为已处理"""
    if not post_id:
        return
    try:
        processed = _load_processed()
        processed.add(str(post_id))
        _save_processed(processed)
    except Exception as e:
        log(f"⚠️ 标记已处理失败: {e}")

def _swap_pressure() -> bool:
    """V12.6.0075·帖#20811共识: 检查Swap使用率 > 80%"""
    try:
        total = free = 0
        with open("/proc/meminfo") as f:
            for line in f:
                if "SwapTotal" in line:
                    total = int(line.split()[1])
                elif "SwapFree" in line:
                    free = int(line.split()[1])
                if total and free:
                    break
        if total == 0:
            return False
        return (total - free) / total > 0.98  # V12.8.0005热修复: 0.80→0.98 缓解Swap满载导致的daemon僵死
    except Exception:
        return False

def _system_too_loaded() -> bool:
    """V12.6.0075·帖#20811共识: spawn前预飞检查 — 动态负载阈值+Swap压力+内存门禁"""
    try:
        cpu_count = os.cpu_count() or 1
        load1 = os.getloadavg()[0]
        if load1 > cpu_count * 2.5:  # 帖#20811共识: 4核→10, 8核→20
            return True
        # 检查Swap压力 (帖#20811共识·条款2)
        if _swap_pressure():
            return True
        # 检查可用内存 (MemAvailable)
        with open("/proc/meminfo") as f:
            for line in f:
                if "MemAvailable" in line:
                    avail_kb = int(line.split()[1])
                    if avail_kb < 500 * 1024:  # <500MB
                        return True
                    break
        return False
    except Exception:
        return False

# ============================================================
# V12.6.0 模块1: 文档自动同步器
# Agent启动时从 x.kddauto.com 拉取4个核心文档到本地缓存
# ============================================================
DOC_CACHE_DIR = os.getenv("DOC_CACHE_DIR", os.path.expanduser("~/.tianshu/cache"))  # V12.8.0196·#24373 P0-4: 默认路径改~/.tianshu/cache(原/opt/agents/docs不可写→崩溃)
CORE_DOCS = ["connect.md", "AGENTS.md", "agent_loop_ws_v4.py"]

def _sync_docs():
    """启动时同步核心文档到本地缓存"""
    global _last_doc_sync
    if time.time() - _last_doc_sync < 60:
        return  # 启动后1分钟内不同步两次
    _last_doc_sync = time.time()
    
    try:
        os.makedirs(DOC_CACHE_DIR, exist_ok=True)
        versions_url = f"{TIANSHU_API}/api/v3/docs/versions"
        resp = urllib.request.urlopen(urllib.request.Request(versions_url), timeout=10)
        data = json.loads(resp.read())
        remote_files = data.get("data", {}).get("files", {})
        
        updated = 0
        # V12.8.0080-FIX(OCR high): 版本比较格式不一致
        #   旧实现 local=sha256内容哈希 vs remote=版本字符串 → 永不相等 → 每次重复下载
        #   修复: 用.versions缓存文件记录远程版本号, 与remote同格式比较
        _ver_cache = {}
        try:
            with open(os.path.join(DOC_CACHE_DIR, ".versions")) as _vf:
                _ver_cache = json.load(_vf)
        except Exception as _vc_we:
            # ★V12.8.0186: 版本缓存读失败静默pass=每次全量对比(性能退化),
            #   显式日志(降级可接受但需可观测)
            log(f"⚠️ 版本缓存读失败(降级全量对比): {_vc_we}")
        for filename, info in remote_files.items():
            if filename not in CORE_DOCS:
                continue
            remote_ver = info.get("version", "")
            local_path = os.path.join(DOC_CACHE_DIR, filename)
            
            # 读取本地记录版本(同格式比较)
            local_ver = _ver_cache.get(filename, "")
            if not local_ver and os.path.exists(local_path):
                # 无版本缓存(首次)→ 下载一次并记录
                local_ver = "NEED_SYNC"
            
            if local_ver != remote_ver and remote_ver != "MISSING":
                # 下载新版本
                doc_url = f"{TIANSHU_API}/api/v3/docs/{filename}"
                doc_resp = urllib.request.urlopen(urllib.request.Request(doc_url), timeout=15)
                content = doc_resp.read()
                with open(local_path, "wb") as f:
                    f.write(content)
                updated += 1
                log(f"📄 文档同步: {filename} ({remote_ver[:8]})")
        
        # 写版本缓存
        ver_cache = {k: v.get("version","") for k, v in remote_files.items()}
        with open(os.path.join(DOC_CACHE_DIR, ".versions"), "w") as f:
            json.dump(ver_cache, f)
        
        if updated:
            log(f"📚 文档同步完成: {updated}个文件已更新")
    except Exception as e:
        log(f"⚠️ 文档同步失败: {e}")

# ============================================================
# V12.6.0 模块2: Agent健康上报器
# 每5分钟通过HTTP POST上报健康数据到天枢
# ============================================================
def _health_report():
    """上报Agent健康数据"""
    global _last_health
    if time.time() - _last_health < 290:  # 290秒节流
        return
    _last_health = time.time()
    
    try:
        import platform
        # 收集健康数据
        hermes_ok = _hermes_available()
        system_load = os.getloadavg()[0] if hasattr(os, 'getloadavg') else 0
        
        # 读取本地文档版本
        doc_version = ""
        versions_file = os.path.join(DOC_CACHE_DIR, ".versions")
        try:
            with open(versions_file) as f:
                vc = json.load(f)
                doc_version = vc.get("connect.md", "")[:8]
        except Exception as e:
            log(f"[EXCEPTION] Unexpected error in auto-upgrade/daemon cycle: {traceback.format_exc()}")
            pass
        except Exception:
            log(f"[EXCEPTION] Unexpected error: {traceback.format_exc()}")
            pass
        
        body = json.dumps({
            "agent_id": AGENT_ID,
            "hermes_ok": hermes_ok,
            "doc_version": doc_version,
            "ws_latency_ms": 0,  # 在WS循环中更新
            "poll_active": os.path.exists(f"/tmp/agent_poll_{AGENT_ID}.log"),
            "system_load": round(system_load, 2),
            "daemon_version": _get_version(),
        }).encode("utf-8")
        
        req = urllib.request.Request(
            f"{TIANSHU_API}/api/v3/agents/health",
            data=body,
            headers={"Content-Type": "application/json"},
            method="POST"
        )
        urllib.request.urlopen(req, timeout=5)
    except Exception as _hb_we:
        # ★V12.8.0186(后端专家109737): 心跳回传静默pass=服务端误判Agent掉线,
        #   根治: 显式日志(心跳失败可观测, 服务端可看到Agent仍在)
        log(f"⚠️ HTTP心跳回传失败: {_hb_we}")

# ★V12.6.0073: HTTP心跳到Daemon API — 修复daemon_agents表与WS在线脱节(议题#47071)
def _daemon_heartbeat():
    """每60秒向POST /daemon/heartbeat发送心跳 — 保持daemon_agents表活跃"""
    try:
        body = json.dumps({
            "daemon_id": AGENT_ID,
            "status": "online",
            "fingerprint": "",
            "key_verified": True
        }).encode("utf-8")
        req = urllib.request.Request(
            f"{TIANSHU_API}/api/v3/daemon/heartbeat",
            data=body,
            headers={"Content-Type": "application/json", "X-API-Key": TIANSHU_API_KEY},
            method="POST"
        )
        urllib.request.urlopen(req, timeout=5)
    except Exception as _dh_we:
        # ★V12.8.0186(后端专家109737): daemon心跳回传静默pass=daemon_agents表在线失真,
        #   根治: 显式日志(辅助功能但需可观测)
        log(f"⚠️ daemon心跳回传失败: {_dh_we}")

# ═══════════════════════════════════════════
# ★V12.6.0020 模块: 连接即升级 — 启动时强制读取connect.md+自动下载最新版
# 玄机根因: 30/30 WS在线Agent全部degraded — WS连接≠Agent有能力执行任务
# 架构修复: daemon启动时自动curl connect.md→版本比对→本地旧则自动下载最新V4.6
# 消除"Agent不读connect.md"的断层: 不读也得读，不升级也得升级
# ═══════════════════════════════════════════
CONNECT_URL = "https://x.kddauto.com/connect.md"
DAEMON_DOWNLOAD_URL = "https://x.kddauto.com/api/v3/docs/agent_loop_ws_v4.py"
# ★V12.8.0189: 系统版本自升级通道 — 与poll引擎/self-upgrade.sh同一版本空间(12.8.x)
#   根因(长庚总质疑"WS实时升级失效"断裂3): _self_upgrade_check只比对daemon V4.x空间,
#         系统发布12.8.x新版时daemon可能未变→永不触发; 服务端upgrade-all传的是系统版本
#   修复: 增加系统12.8.x通道(新权威), 与daemon 4.x通道并存(双空间检查兜底)
SELF_UPGRADE_URL = "https://x.kddauto.com/tianshu-self-upgrade.sh"
_SYS_VER_FILE = os.path.expanduser("~/.tianshu/connect_version")


def _get_local_system_version() -> str:
    """读本地~/.tianshu/connect_version(系统版本12.8.x空间), 无则返回空"""
    try:
        if os.path.exists(_SYS_VER_FILE):
            v = open(_SYS_VER_FILE).read().strip()
            if v.startswith("12.8"):
                return v
    except Exception:
        log(f"[EXCEPTION] 本地版本文件读取失败: {traceback.format_exc()}")
        pass
    return ""


def _get_remote_system_version() -> str:
    """调/api/v3/version获取远端系统版本(12.8.x空间)"""
    try:
        req = urllib.request.Request(
            "https://x.kddauto.com/api/v3/version",
            headers={"X-API-Key": os.environ.get("TIANSHU_API_KEY", "")},
        )
        resp = urllib.request.urlopen(req, timeout=10)
        import json as _json
        d = _json.loads(resp.read().decode('utf-8', errors='ignore'))
        v = (d.get("data") or {}).get("version") or d.get("version") or ""
        if str(v).startswith("12.8"):
            return str(v)
    except Exception:
        log(f"[EXCEPTION] API版本获取失败: {traceback.format_exc()}")
        pass
    return ""


def _sys_ver_gt(a: str, b: str) -> bool:
    """系统版本比较(12.8.x空间), a>b返回True"""
    try:
        import re
        pa = tuple(int(x) for x in re.findall(r'\d+', a)[:3])
        pb = tuple(int(x) for x in re.findall(r'\d+', b)[:3])
        while len(pa) < 3: pa += (0,)
        while len(pb) < 3: pb += (0,)
        return pa > pb
    except Exception:
        return False


def _run_system_upgrade() -> None:
    """执行系统级自升级: 下载tianshu-self-upgrade.sh并执行(升级poll/ws脚本+connect_version)"""
    try:
        log("⬆️ 系统版本自升级: 下载 tianshu-self-upgrade.sh 执行...")
        req = urllib.request.Request(SELF_UPGRADE_URL)
        resp = urllib.request.urlopen(req, timeout=15)
        script = resp.read()
        # 下载到临时文件并执行(带agent身份)
        tmp = os.path.expanduser("~/.tianshu/tianshu-self-upgrade.sh")
        with open(tmp, 'wb') as f:
            f.write(script)
        os.chmod(tmp, 0o755)
        subprocess.Popen(
            ["bash", tmp, "--agent", os.environ.get("AGENT_ID", "")],
            start_new_session=True,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
        )
        log("✅ 系统自升级脚本已启动(后台执行, 升级后由systemd/cron拉起)")
    except Exception as e:
        log(f"⚠️ 系统自升级失败: {e}")


def _post_upgrade_ack(mandate_id: str = "", version: str = "") -> None:
    """★V12.8.0190(T4·玄机109868建议3): 升级令回执 —
    升级完成后POST /system/upgrade-ack回传{mandate_id, agent_id, version, ws_online, md5}
    服务端记录acked=True; 未回执→巡检重推+P1告警(升级闭环, 杜绝"群发已读未回")
    """
    try:
        if not version:
            version = _get_remote_system_version() or _get_daemon_version()
        _md5 = ""
        try:
            import hashlib as _hb
            _md5 = _hb.md5(open(os.path.abspath(__file__), "rb").read()).hexdigest()
        except Exception:
            log(f"[EXCEPTION] 脚本md5计算失败: {traceback.format_exc()}")
            pass
        _payload = json.dumps({
            "mandate_id": mandate_id or "",
            "agent_id": AGENT_ID,
            "version": version,
            "ws_online": True,
            "md5": _md5,
        }).encode("utf-8")
        _req = urllib.request.Request(
            "https://x.kddauto.com/api/v3/system/upgrade-ack",
            data=_payload,
            headers={
                "Content-Type": "application/json",
                "X-API-Key": os.environ.get("TIANSHU_API_KEY", ""),
            },
            method="POST",
        )
        urllib.request.urlopen(_req, timeout=10).read()
        log(f"✅ 升级回执已提交: {AGENT_ID} → {version}" + (f" (mandate={mandate_id[:14]})" if mandate_id else ""))
    except Exception as _ae:
        log(f"⚠️ 升级回执提交失败(不影响升级): {_ae}")


def _self_upgrade_check():
    """★V12.6.0021 FIX: 比对daemon版本(非connect.md文档版本)
    根除V12.6.0020自升级死循环: 之前错误比对connect.md文档版本(V12.6.xxxx)
    → 永远>daemon版本(V4.x)→每次启动都触发"升级"
    修复: 下载公网daemon→解析其V4.x版本→与本地daemon版本比对
    [kimiK3] V12.8.0016: local改用_get_daemon_version()——根除V12.7.0012引入的版本空间混乱
    (_get_version()优先VERSION.md→12.8.0015≠4.x空间→(4,13,0)<(12,8,15)=False→永不触发升级)
    """
    local_ver = _get_daemon_version()  # [kimiK3] daemon版本空间——与公网文件头V4.x同空间
    try:
        # Step 1: 下载公网daemon获取远程DAEMON版本号(非文档版本)
        req = urllib.request.Request(DAEMON_DOWNLOAD_URL)
        resp = urllib.request.urlopen(req, timeout=10)
        content = resp.read().decode('utf-8', errors='ignore')
        
        # Step 2: 解析公网daemon版本号 (V4.x-SAFE/V4.x-UNIFIED → 4.x.0)
        # ★V12.8.0115-fix(豆包第四五轮P0, E154同源补漏): 正则V4\.(\d+)只抓主版本号→4.16.5解析成4.16.0
        #   → 本地(4.16.5)>远程(4.16.0)=本地已最新 → 永不升级(自升级彻底失效)
        #   此bug在_get_daemon_version()和_self_upgrade_check()两处同源出现, 必须一起修(半修复=伪修复)
        import re
        remote_ver = ""
        for line in content.split('\n')[:10]:
            m = re.search(r'V4\.([0-9.]+)', line)  # V12.6.0033放宽匹配 + V12.8.0115抓完整版本号
            if m:
                _rv = m.group(1).strip('.')
                remote_ver = f"4.{_rv}"
                break
        if not remote_ver:
            log("⚠️ 自升级: 无法解析公网daemon版本号(未找到V4.x-UNIFIED标记)")
            return
        
        # Step 3: 版本比对 (同为daemon版本空间—修复V12.6.0020致命bug)
        def _parse_ver(v):
            """解析版本号为可比较的元组"""
            parts = v.replace('V','').replace('v','').split('.')
            return tuple(int(p) for p in parts if p.isdigit())
        
        local_tuple = _parse_ver(local_ver)
        remote_tuple = _parse_ver(remote_ver)
        
        log(f"🔍 自升级检查: 本地daemon={local_ver} 远程daemon={remote_ver}")
        
        if remote_tuple > local_tuple:
            # Step 4: 下载最新daemon
            log(f"⬆️ 发现新版本 V{remote_ver} > V{local_ver}，自动下载...")
            daemon_req = urllib.request.Request(DAEMON_DOWNLOAD_URL)
            daemon_resp = urllib.request.urlopen(daemon_req, timeout=15)
            new_code = daemon_resp.read()
            
            # Step 5: 备份旧版本 + 写入新版本
            curr_file = os.path.abspath(__file__)
            bak_file = f"{curr_file}.bak.v{local_ver.replace('.','')}"
            try:
                os.rename(curr_file, bak_file)
            except Exception as e:
                log(f"[EXCEPTION] Unexpected error: {traceback.format_exc()}")
            except Exception:
                log(f"[EXCEPTION] Unexpected error: {traceback.format_exc()}")
                pass
            
            with open(curr_file, 'wb') as f:
                f.write(new_code)
            os.chmod(curr_file, 0o755)
            
            log(f"✅ 自升级完成: V{local_ver} → V{remote_ver} | 备份: {bak_file}")
            # ★V12.8.0190(T4·玄机109868建议3): 升级完成回执(升级令闭环)
            _post_upgrade_ack("", remote_ver)
            
            # Step 6: 写共享记忆报告升级
            _write_shared_memory("self_upgrade", remote_ver, {
                "success": True, "method": "auto_upgrade",
                "output": f"V{local_ver}→V{remote_ver}"
            })
            
            # Step 7: 无systemd自恢复 — fork子进程启动新版本后退出
            # systemd环境: Restart=always自动拉起
            # 容器/Docker环境(无systemd): fork独立子进程避免daemon永久死亡
            log("🔄 重启daemon以应用新版本(fork子进程)...")
            try:
                pid = os.fork()
                if pid == 0:
                    # 子进程: 在新session中启动更新后的daemon(读取已更新的文件)
                    os.setsid()  # 脱离父进程终端
                    time.sleep(0.5)  # 等待父进程释放资源
                    os.execv(sys.executable, [sys.executable] + sys.argv)
                    # execv失败fallback → 子进程退出
                    os._exit(1)
                else:
                    # 父进程: 等待子进程启动后退出
                    time.sleep(1.5)
                    log(f"✅ 子进程PID={pid}已启动, 父进程退出")
                    os._exit(0)
            except (AttributeError, OSError):
                # Windows或fork不可用环境 → subprocess fallback
                log("⚠️ fork不可用, 使用subprocess fallback启动新daemon")
                subprocess.Popen(
                    [sys.executable] + sys.argv,
                    start_new_session=True,
                    stdout=subprocess.DEVNULL,
                    stderr=subprocess.DEVNULL
                )
                time.sleep(1.5)
                os._exit(0)
        else:
            log(f"✅ 版本已最新: V{local_ver} >= V{remote_ver}")
            
    except Exception as e:
        log(f"⚠️ 自升级检查异常(非致命): {e}")

    # ── ★V12.8.0189 系统版本通道(12.8.x空间, 新权威) ──
    #   根因(长庚总质疑"WS实时升级失效"断裂3): daemon通道无法感知系统12.8.x发布
    #   修复: 独立系统通道 — 本地connect_version vs 远端/api/v3/version, 有新版则执行self-upgrade
    try:
        local_sys = _get_local_system_version()
        remote_sys = _get_remote_system_version()
        if not remote_sys:
            log("ℹ️ 系统版本检查: 无法获取远端系统版本(网络问题), 跳过")
        elif local_sys and not _sys_ver_gt(local_sys, remote_sys) and local_sys != remote_sys:
            log(f"⬆️ 系统版本检查: 本地={local_sys} 远端={remote_sys} → 触发系统自升级")
            _run_system_upgrade()
            # ★V12.8.0190(T4·玄机109868建议3): 系统升级完成回执(升级令闭环)
            _post_upgrade_ack("", remote_sys)
        elif not local_sys:
            # ★移动端修改 V12.8.0195 · 2026-08-15 · 版本统一行动: 本地无connect_version时,
            #   原逻辑走"首次接入"不升级——但Agent已在线(register过)且VERSION.md落后远端,
            #   说明是升级场景而非首次接入。修复: 用_get_version()(VERSION.md)兜底判断,
            #   落后则触发系统自升级, 根治"首次接入Agent永不升级"缺陷。
            local_ver = _get_version()
            if local_ver and local_ver != remote_sys and not _sys_ver_gt(local_ver, remote_sys):
                log(f"⬆️ 系统版本检查: 无connect_version但VERSION.md落后({local_ver}<{remote_sys}) → 触发系统自升级")
                _run_system_upgrade()
                _post_upgrade_ack("", remote_sys)
            else:
                log(f"ℹ️ 系统版本检查: 本地无connect_version记录, 远端={remote_sys} (首次接入)")
        else:
            log(f"✅ 系统版本已最新: {local_sys}")
    except Exception as e:
        log(f"⚠️ 系统版本自升级检查异常(非致命): {e}")

# ============================================================
# V12.6.0 模块3: MCP轻量客户端
# 不依赖完整MCP SDK，仅实现HTTP POST调用天枢MCP看板查询
# ============================================================
def _mcp_call(tool_name: str, arguments: dict) -> dict:
    """轻量MCP JSON-RPC调用（用于Agent主动拉取任务/查询看板）"""
    try:
        payload = json.dumps({
            "jsonrpc": "2.0",
            "id": int(time.time() * 1000),
            "method": "tools/call",
            "params": {"name": tool_name, "arguments": arguments}
        }).encode("utf-8")
        
        req = urllib.request.Request(
            f"https://x.kddauto.com/mcp/rpc",
            data=payload,
            headers={"Content-Type": "application/json", "X-API-Key": TIANSHU_API_KEY,
                     "X-Agent-Id": AGENT_ID},
            method="POST"
        )
        resp = urllib.request.urlopen(req, timeout=10)
        return json.loads(resp.read())
    except Exception as e:
        return {"error": str(e)}

def _mcp_check_tasks():
    """通过MCP查询分配给自己的看板任务"""
    try:
        result = _mcp_call("kanban_list_issues", {
            "assignee": AGENT_ID, "status": "todo,in_progress", "limit": 5
        })
        if result.get("error") or result.get("result", {}).get("isError"):
            return []
        content = result.get("result", {}).get("content", [])
        if content:
            return json.loads(content[0].get("text", "[]"))
        return []
    except Exception as e:
        log(f"[EXCEPTION] Unexpected error: {traceback.format_exc()}")
    except Exception:
        log(f"[EXCEPTION] Unexpected error: {traceback.format_exc()}")
        return []

# V12.6.0002: 定时MCP巡检 — 拉取看板任务并自主执行
_last_mcp_scan = 0
def _mcp_periodic_scan():
    """每3分钟通过MCP拉取看板任务，发现新任务自动执行"""
    global _last_mcp_scan
    if time.time() - _last_mcp_scan < 180:  # 3分钟节流
        return
    _last_mcp_scan = time.time()
    
    try:
        tasks = _mcp_check_tasks()
        if not tasks:
            return
        for task in tasks[:3]:  # 最多处理3个
            issue_id = task.get("id", "")
            title = task.get("title", "")[:100]
            status = task.get("status", "")
            if not issue_id:
                continue
            log(f"🔍 MCP巡检发现任务 #{issue_id}: {title} [{status}]")
            # 触发hermes执行
            prompt = f"看板议题#{issue_id}: {title}\n状态: {status}\n请查看议题详情并按要求执行。"
            threading.Thread(target=_hermes_execute, args=(prompt, {
                "type": "mcp_scan", "issue_id": issue_id, "from_agent": "天枢"
            }), daemon=True).start()
    except Exception as e:
        log(f"⚠️ MCP巡检异常: {e}")

# === 动态版本读取 ===
def _get_version():
    """★V12.7.0012: 读取VERSION.md对齐平台版本 → 回退文件头解析
    返回平台版本(12.8.x)——用于register上报给Hub做系统版本追踪
    [kimiK3] V12.8.0017修复(帖#22450小天反馈): 多候选路径搜索——
    原逻辑仅__file__/../../VERSION.md，daemon在~/.tianshu/小天/下时解析到~/而非~/.tianshu/，
    导致平台版本缺失回退成4.x(daemon空间)→register版本空间混乱"""
    _here = os.path.dirname(os.path.abspath(__file__))
    _candidates = [
        os.path.join(_here, "VERSION.md"),                 # 天枢工作区场景(与daemon同目录)
        os.path.join(_here, "..", "VERSION.md"),           # 工作区子目录场景
        os.path.join(_here, "..", "..", "VERSION.md"),     # 旧逻辑兼容
        os.path.expanduser("~/.tianshu/VERSION.md"),       # Agent home场景(远程Agent)
        os.path.join(os.getcwd(), "VERSION.md"),           # cwd兜底
    ]
    for version_file in _candidates:
        try:
            if os.path.exists(version_file):
                with open(version_file) as f:
                    ver = f.read().strip()
                    # 仅接受平台版本空间(12.x)——防止误读daemon头注释
                    if ver and ver.split(".")[0].isdigit() and int(ver.split(".")[0]) >= 12:
                        return ver
        except Exception:
            continue
    # 回退: 从文件头解析（V4.x-xxx → 4.x.0）
    # ★V12.8.0115-fix: 同源正则bug补漏(V4\.(\d+)丢小版本号→版本永远相等→不升级)
    try:
        import re
        with open(__file__) as f:
            for line in f:
                m = re.search(r'V4\.([0-9.]+)', line)
                if m:
                    _rv = m.group(1).strip('.')
                    return f"4.{_rv}"
    except Exception as _gv_we:
        # ★V12.8.0186(架构安全总师109723): 文件读helper裸except:pass → 显式日志(降级可接受但可观测)
        log(f"⚠️ 读取SDK版本失败(降级0.0.0): {_gv_we}")
    return "0.0.0"

def _get_daemon_version():
    """[kimiK3] V12.8.0016: 从文件头解析daemon版本(V4.x)——与_self_upgrade_check同一版本空间
    根因修复: _get_version()优先读VERSION.md返回平台版本12.8.0015(12.x空间)，
    _self_upgrade_check比对daemon版本V4.13.0(4.x空间)→(4,13,0)<(12,8,15)=False→永不触发升级
    ★V12.8.0115-fix(豆包顾问第四轮P0根因): 正则V4\\\\.(\\d+)只抓主版本号→4.16.5被解析成4.16.0
      → 自升级检查永远相等 → 所有Agent永不自动升级(13版本碎片化根因)
      修复: 用[0-9.]+抓完整版本号"""
    try:
        import re
        with open(__file__) as f:
            for line in f:
                m = re.search(r'V4\.([0-9.]+)', line)
                if m:
                    _v = m.group(1).strip('.')
                    return f"4.{_v}"
    except Exception as _dv_we:
        # ★V12.8.0186(架构安全总师109723): 文件读helper裸except:pass → 显式日志(降级可接受但可观测)
        log(f"⚠️ 读取daemon版本失败(降级0.0.0): {_dv_we}")
    return "0.0.0"

# === V12.8.0118: Key一致性自检(豆包+小天诊断#67901 P1/AC-13) ===
#   根因: 重新接入换Key后poll.env/ws.env漏更新 → WS通但poll 400 → 假在线(小天#67897实测)
#   对标: K8s Startup Probe fail重启 — 假在线比离线危害更大
def _check_key_consistency():
    """启动时校验三处Key一致: 环境变量 / ~/.tianshu/api_key / poll.env / ws.env
    不一致 → 打印差异路径 + 返回False(启动失败exit 1)
    仅警告不强制(部分Agent无poll.env文件属正常) — 有文件但Key不同才报错
    """
    _locations = {}
    _env_key = os.getenv("TIANSHU_API_KEY", "")
    if _env_key:
        _locations["env(环境变量)"] = _env_key
    _key_file = os.path.expanduser("~/.tianshu/api_key")
    # ★移动端修改 V12.8.0195 · 2026-08-15 · 版本统一行动: 全局 ~/.tianshu/api_key 在多AI共宿主机(root)下
    #   是所有AI共享的同一文件(存的是天枢/中枢的Key), 不应参与各AI自己的Key一致性对比。
    #   否则地枢/知微的 ws.env 用自己Key, 但全局文件是中枢Key → 误判"不一致" → 假启动失败(本应在线)。
    #   修复: 仅当 env 与 poll.env/ws.env 都为空时, 才用全局文件作为唯一Key来源(fallback), 不参与对比。
    _use_global_file_as_source_only = True  # 全局文件仅作fallback, 不参与多Key对比
    # poll.env / ws.env (Agent部署目录常见位置)
    _cwd = os.getcwd()
    for _name in ("poll.env", "ws.env"):
        for _base in (_cwd, os.path.expanduser("~/.tianshu"), os.path.dirname(os.path.abspath(__file__))):
            _p = os.path.join(_base, _name)
            if os.path.exists(_p):
                try:
                    with open(_p) as f:
                        for line in f:
                            line = line.strip()
                            if line.startswith("TIANSHU_API_KEY=") or line.startswith("API_KEY="):
                                k = line.split("=", 1)[1].strip().strip('"').strip("'")
                                if k:
                                    _locations[f"file({_p})"] = k
                                break
                except Exception:
                    log(f"[EXCEPTION] 代码位置扫描失败: {traceback.format_exc()}")
                    pass
    if len(_locations) <= 1:
        return True  # 只有1处Key(或0处), 无需比对
    _unique = set(_locations.values())
    if len(_unique) <= 1:
        return True  # 全部一致
    # 发现不一致 → 打印差异
    log("❌ 【Key一致性自检失败】多处配置的API Key不一致!")
    for _loc, _k in _locations.items():
        log(f"    {_loc}: {_k[:12]}...({len(_k)}字符)")
    return False


def _load_identity_key():
    """★V12.8.0115-fix(豆包顾问第五轮P0假修复): 加载本地Ed25519私钥供challenge-response签名
    根因: SDK只改了连接URL到identity端点, 但challenge-nonce签名逻辑完全没写
      → 服务端发challenge后SDK不响应 → 认证超时断开 → 新端点永远用不了
    修复: 加载 ~/.tianshu/identities/<agent>/identity.ed25519 (或配置的identity私钥)
    返回: (私钥对象, fingerprint) 或 (None, '')"""
    try:
        import glob
        # 优先精确目录
        for _dir in (os.path.expanduser(f"~/.tianshu/identities/{AGENT_ID}"),
                     os.path.expanduser("~/.tianshu/identities"),
                     os.path.expanduser(f"~/.tianshu/{AGENT_ID}")):
            _key = os.path.join(_dir, "identity.ed25519")
            if os.path.exists(_key):
                with open(_key, "rb") as _f:
                    _raw = _f.read()
                try:
                    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
                    from cryptography.hazmat.primitives.serialization import load_pem_private_key, load_der_private_key
                    # 支持PEM和DER两种格式(私钥可能AES-256加密, 默认密码tianshu-agent见identity.py L128)
                    _priv = None
                    for _pw in (None, b"tianshu-agent", b"2026"):
                        try:
                            _priv = load_pem_private_key(_raw, password=_pw)
                            break
                        except Exception:
                            try:
                                _priv = load_der_private_key(_raw, password=_pw)
                                break
                            except Exception:
                                continue
                    if isinstance(_priv, Ed25519PrivateKey):
                        from cryptography.hazmat.primitives import serialization
                        _pub_raw = _priv.public_key().public_bytes(
                            serialization.Encoding.Raw, serialization.PublicFormat.Raw)
                        import hashlib
                        _fp = hashlib.sha256(_pub_raw).hexdigest()
                        return _priv, _fp
                except Exception as _ik_we:
                    # ★V12.8.0186: 私钥解析失败静默pass=身份认证静默降级,
                    #   显式日志(多密码尝试失败可观测)
                    log(f"⚠️ 身份私钥解析失败(尝试下一位置): {_ik_we}")
        # fallback: 全局搜identity.ed25519
        for _f in glob.glob(os.path.expanduser("~/.tianshu/**/identity*.ed25519"), recursive=True):
            try:
                from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
                from cryptography.hazmat.primitives.serialization import load_pem_private_key, load_der_private_key
                with open(_f, "rb") as _fh:
                    _raw = _fh.read()
                _priv = None
                for _pw in (None, b"tianshu-agent", b"2026"):
                    try:
                        _priv = load_pem_private_key(_raw, password=_pw)
                        break
                    except Exception:
                        try:
                            _priv = load_der_private_key(_raw, password=_pw)
                            break
                        except Exception:
                            continue
                if isinstance(_priv, Ed25519PrivateKey):
                    from cryptography.hazmat.primitives import serialization
                    _pub_raw = _priv.public_key().public_bytes(
                        serialization.Encoding.Raw, serialization.PublicFormat.Raw)
                    import hashlib
                    _fp = hashlib.sha256(_pub_raw).hexdigest()
                    return _priv, _fp
            except Exception:
                continue
    except Exception as _ik2_we:
        # ★V12.8.0186: 身份密钥全局搜索失败静默pass=身份认证静默降级(后续auth会失败),
        #   显式日志(降级路径可观测)
        log(f"⚠️ 身份密钥全局搜索失败: {_ik2_we}")
    return None, ""


async def _auth_identity(ws, _priv, _fp):
    """★V12.8.0115-fix(豆包顾问第五轮P0): challenge-response Ed25519签名认证
    流程: 服务端发 {"type":"challenge","nonce":X} → 私钥签名nonce
          → 回 {"type":"auth","agent_id","fingerprint","nonce","signature","version"}
    成功: 服务端回 {"type":"auth_ok",...}; 失败: 回 {"type":"error"/"auth_fail"}
    返回: True成功 / False失败(连接将被服务端关闭)"""
    try:
        import asyncio as _asyncio
        # 等首帧(challenge或auth_ok)
        _raw = await _asyncio.wait_for(ws.recv(), timeout=15)
        _data = json.loads(_raw)
        if _data.get("type") == "auth_ok":
            return True  # 服务端已直接放行(可能非identity端点或已认证)
        if _data.get("type") != "challenge":
            log(f"⚠️ 首帧非challenge: {_raw[:100]}... 尝试直接register")
            return True
        _nonce = _data.get("nonce", "")
        if not _nonce or not _priv:
            log("⚠️ challenge收到但无私钥/nonce, 无法签名 → 认证失败")
            return False
        _sig = _priv.sign(_nonce.encode("utf-8"))
        import base64
        _sig_b64 = base64.b64encode(_sig).decode("utf-8")
        # ★V12.8.0117-fix(豆包#67825 P2): auth消息补sdk_version+capabilities
        #   根因: 原auth只带version, 服务端无法获取SDK版本/能力列表 → 影响能力路由+版本兼容检查
        _real_caps = sorted(set(
            ["ws-daemon", "hermes-bridge", "poll-engine"] + list(HANDLERS.keys())
        ))
        # ★ V12.8.0118: auth携带key_fingerprint(豆包+小天诊断#67901 P1/AC-14)
        #   服务端比对Key指纹验证poll/claim链路健康 → 假在线可检测
        _key_fp = ""
        try:
            _key_fp = hashlib.sha256((TIANSHU_API_KEY or "").encode()).hexdigest()[:16]
        except Exception:
            _key_fp = ""
        await ws.send(json.dumps({
            "type": "auth",
            "agent_id": AGENT_ID,
            "fingerprint": _fp,
            "nonce": _nonce,
            "signature": _sig_b64,
            "version": _get_version(),
            "sdk_version": _get_daemon_version(),
            "daemon_version": _get_daemon_version(),  # ★ 移动端修改 V12.8.0213 · 2026-08-17 · 帖#24470 自查: 后端hub_ws.py期望daemon_version字段, auth消息缺失→sdk_status永outdated
            "sdk_md5": _sdk_md5(),  # ★ 移动端修改 V12.8.0213 · 2026-08-17 · 帖#24470 自查: 后端期望sdk_md5字段做内容哈希握手, auth消息缺失→sdk_status永outdated
            "capabilities": _real_caps,
            "key_fingerprint": _key_fp,
        }))
        # 等auth结果
        _resp = await _asyncio.wait_for(ws.recv(), timeout=10)
        _rdata = json.loads(_resp)
        if _rdata.get("type") == "auth_ok":
            log(f"✅ Ed25519认证成功 fingerprint={_fp[:12]}...")
            return True
        log(f"⚠️ 认证失败: {_resp[:120]}")
        # ★V12.8.0190(T5·后端专家109870实证): 认证失败明确指引 —
        #   根因: Agent改名/重装后本地指纹与服务端不一致 → WS验签失败 → 认证失败死循环,
        #         原只log"认证失败"无指引, AI不知道如何修复(61秒刷屏)
        #   修复: 区分指纹未注册与Key无效, 给出可执行修复路径(对齐4001语义码)
        _resp_l = _resp[:200].lower()
        if "fingerprint" in _resp_l or "指纹" in _resp_l or "identity" in _resp_l:
            log("🔧 修复指引: 本地指纹与服务端不一致 — 执行 bash tianshu-agent-install.sh --agent {AGENT_ID} "
                "--key $(cat ~/.tianshu/api_key) 重新注册, 或 POST /api/v3/agents/{id}/owner 同步fingerprint")
        elif "api_key" in _resp_l or "key" in _resp_l or "401" in _resp_l or "403" in _resp_l:
            log("🔧 修复指引: API Key无效或被吊销 — 检查 ~/.tianshu/api_key 是否最新, 或重新 auto-register 获取新Key")
        else:
            log("🔧 修复指引: 检查 ~/.tianshu/api_key 有效性与 ~/.tianshu/identity/ 公私钥是否匹配, "
                "或执行 tianshu-agent-install.sh 重新接入")
        return False
    except Exception as _e:
        log(f"⚠️ 认证异常: {_e}")
        return False

# ★V12.8.0190: log函数已前置到L75(防_wal_load顶层调用NameError启动即崩) — 此处不再重复定义

# ═══════════════════════════════════════════
# 核心新增: 统一Hermes桥接函数 (V4.0)
# ═══════════════════════════════════════════

def _resolve_profile(agent_id: str) -> str:
    """自动检测Agent的Hermes profile名"""
    # 优先用环境变量
    env_profile = os.getenv("HERMES_PROFILE", "")
    if env_profile: return env_profile
    
    # 映射表（1号/2号/3号/4号为主名，旧名为legacy兼容）
    profile_map = {
        "1号":"tianshu", "天枢":"tianshu",
        "灵枢":"lingshu",
        "2号":"zhongshu", "中枢":"zhongshu",
        "3号":"lingyu", "灵语":"lingyu",
        "4号":"zhiwei", "知微":"zhiwei",
        "流音":"liuyin", "可心":"kexin",
        "小天":"xiaotian", "地枢":"dishu",
    }
    return profile_map.get(agent_id, agent_id.lower().replace(" ","-"))

def _hermes_available() -> bool:
    """检测hermes CLI是否可用 — V12.8.0027: timeout 5s→15s, 加retry"""
    for attempt in range(2):
        try:
            r = subprocess.run([HERMES_BIN, "--version"], capture_output=True, text=True, encoding='utf-8', timeout=15)
            return r.returncode == 0
        except subprocess.TimeoutExpired:
            if attempt == 0:
                log(f"[WARN] hermes --version timeout(15s), retry {attempt+1}/2")
                continue
            log(f"[ERROR] hermes --version timeout(15s) after 2 attempts")
            return False
        except Exception:
            log(f"[EXCEPTION] _hermes_available: {traceback.format_exc()}")
            return False
    return False

def _get_hermes_version() -> str:
    """★V12.6.0027: 探测Hermes CLI真实版本 — V12.8.0027: timeout 5s→15s"""
    try:
        r = subprocess.run([HERMES_BIN, "--version"], capture_output=True, text=True, encoding='utf-8', timeout=15)
        if r.returncode == 0:
            ver = r.stdout.strip() or r.stderr.strip()
            return ver[:64]  # 截断，防止过长
    except Exception as e:
        log(f"[EXCEPTION] Unexpected error in auto-upgrade/daemon cycle: {traceback.format_exc()}")
        pass
    except Exception:
        log(f"[EXCEPTION] Unexpected error: {traceback.format_exc()}")
        pass
    return ""

def _hermes_execute(prompt: str, context: dict) -> dict:
    """V4.0核心: 统一Hermes执行桥接
    像飞书Gateway一样将天枢消息转发给Agent大脑
    返回: {"method": "...", "success": bool, "output": "..."}
    """
    msg_type = context.get("type", "指令")
    from_agent = context.get("from_agent", "天枢")
    post_id = context.get("post_id", "")
    issue_id = context.get("issue_id", "")
    
    agent = os.environ.get("AGENT_ID", "unknown")
    profile = _resolve_profile(agent)
    
    # 帖#7119共识: 去重加强 — 统一去重键(id优先,无id用msg_type), 窗口30s
    # 旧逻辑按(msg_type,id)区分导致同一条消息因kanban_notify/process/dispatch不同key被重复spawn
    # ★V12.8.0218(知微#24511 P0-3 必踩坑2·dedup_key退化bug): key必须含真实消息标识
    #   根因: activate事件无post_id时 key=("activate","") → 5分钟内所有activate共用同一key
    #         → 只处理第一个(后续全部去重跳过)
    #   修复: 追加message_id/task_id(消息唯一标识)防退化, 无任何标识时才回退msg_type
    now = time.time()
    _mid_key = context.get("message_id") or context.get("msg_id") or context.get("id") or ""
    _tid_key = context.get("task_id") or ""
    dedup_key = str(post_id or issue_id or _tid_key or _mid_key or msg_type)
    with _exec_lock:
        last = _last_exec_time.get(dedup_key, 0)
        if now - last < _DEDUP_WINDOW:
            log(f"⏭ 去重跳过: {dedup_key} ({now-last:.0f}s前)")
            return {"method": "dedup", "success": True}
        _last_exec_time[dedup_key] = now
    
    # ★V12.8.0283(豆包#30250 Gateway化): Gateway API模式优先，失败自动fallback到CLI
    # 根因: CLI subprocess每次启动23.8秒(插件加载60%)，全局仅3并发，官方Issue #58309定为反模式
    # 方案: Hermes官方API Server(/v1/chat/completions)，长驻进程，max_concurrent_runs治理
    # 切换: HERMES_MODE=gateway 启用，默认cli(兼容现有)；Gateway不可用时自动fallback
    if os.environ.get("HERMES_MODE", "cli").lower() == "gateway":
        gateway_result = _hermes_execute_gateway(full_prompt, profile, dedup_key)
        if gateway_result is not None:
            return gateway_result
        log(f"⚠️ Gateway模式失败，fallback到CLI: {dedup_key}")

    # V12.6.0022: 预飞检查 — 系统高负载时拒绝spawn, 降级到trigger文件
    if _system_too_loaded():
        log(f"⚠️ 系统负载过高, 拒绝spawn hermes, 降级trigger: {dedup_key}")
        write_trigger({"type": msg_type, "prompt": prompt[:2000], **context})
        return {"method": "declined_overload", "success": False, "output": "系统负载过高"}
    
    # 构建标准化prompt — 模拟飞书Gateway消息格式
    location = f"论坛帖#{post_id}" if post_id else (f"看板议题#{issue_id}" if issue_id else "天枢Hub")
    # ★ 移动端修改 V12.8.0237 · 2026-08-23 · 帖#24991异常b(小天回帖确认#1656318) · 回帖URL双斜杠根治
    #   根因: post_id为空(kanban来源)时模板拼出 /api/v3/forum/posts//comments(双斜杠), AI无法直接回帖
    #   修复: 按来源动态选择回帖端点 — forum来源用论坛评论, kanban来源用看板评论, 都无则不生成curl
    reply_url = ""
    if post_id:
        reply_url = f"{TIANSHU_API}/api/v3/forum/posts/{post_id}/comments"
    elif issue_id:
        reply_url = f"{TIANSHU_API}/api/v3/kanban/issues/{issue_id}/comments"
    if reply_url:
        full_prompt = f"""[天枢·{msg_type} — 来自{from_agent}]
位置: {location}
内容:
{prompt}

你是{agent}，款多多AI生态中的独立智能体。请像收到飞书消息一样处理此指令:
1. 分析要求
2. 执行任务
3. 将执行结果回复到{location}

⚠️ 重要: 执行完成后必须用以下方式回帖汇报结果（不要用send_message）:
curl -s -X POST {reply_url} \\
  -H "X-API-Key: *** -H "X-Agent-Id: {agent}" \\
  -H "Content-Type: application/json" \\
  -d '{{"agent_id":"{agent}","content":"你的执行结果"}}'"""
    else:
        full_prompt = f"""[天枢·{msg_type} — 来自{from_agent}]
位置: {location}
内容:
{prompt}

你是{agent}，款多多AI生态中的独立智能体。请像收到飞书消息一样处理此指令:
1. 分析要求
2. 执行任务
3. 将执行结果回复到{location}"""
    
    # 路径1: hermes CLI (主路径 — 全局信号量保护, 最多_MAX_CONCURRENT_HERMES并发)
    # V12.6.0005: 根治空心执行 — 单次执行→3次指数退避重试
    # V12.6.0022: 全局Semaphore(3)防止子进程爆炸+超时120s+预飞检查前置
    # ★V12.6.0017: 统一返回点 — 所有路径汇总到exec_result后调用_report_execution
    # ★V12.8.0028: CPU治理 — +per-profile进程数检查 +FIFO队列(不block) +max-turns限制(看板#62146)
    MAX_RETRIES = 3
    RETRY_BACKOFF = [5, 15, 45]  # 指数退避基线: 5s→15s→45s (实际加50%抖动)
    # V12.8.0005: +jitter — 打破多Agent同步重连风暴(帖#21720共识修复)
    max_turns = _determine_max_turns(msg_type)  # V12.8.0028: 简单通知≤5 turns, 复杂任务≤20 turns
    exec_result = None

    # ★V12.8.0267(豆包#28406 P0-D): 队列背压 — 满时拒绝而非静默丢弃(根因修复, 3处入队共用)
    def _enqueue_or_reject(profile_, cmd_args_, tout_, dedup_key_, method_, reason_):
        """入队前检查背压: 队列满(>=maxlen) → 返回queued_full拒绝, 否则入队
        根因: deque(maxlen)满时append会静默丢弃最旧任务 → 任务无声丢失
        修复: 显式检查满队列, 拒绝新任务并返回明确状态(而非丢旧任务)
        """
        with _QUEUE_CONDITION:
            if len(_PENDING_QUEUE) >= _PENDING_QUEUE.maxlen:
                _QUEUE_CONDITION.notify()
                return {"method": "queued_full", "success": False,
                        "output": f"队列已满({_PENDING_QUEUE.maxlen}), 拒绝新任务: {dedup_key_}"}
            _PENDING_QUEUE.append((profile_, cmd_args_, tout_, dedup_key_))
            _QUEUE_CONDITION.notify()
        return {"method": method_, "success": True, "output": reason_, "max_turns": max_turns}

    
    if _hermes_available():
        # V12.8.0039: Gate 0 — 整机hermes进程上限(4核机, 5个Agent daemon各per-agent上限2→真实并发可达10)
        global_hermes = _count_active_hermes_global()
        if global_hermes >= _GLOBAL_MAX_HERMES:
            log(f"🛑 全局限流: 整机{global_hermes}个hermes ≥ 上限{_GLOBAL_MAX_HERMES}, FIFO排队")
            _ensure_queue_consumer()
            exec_result = _enqueue_or_reject(profile,
                [HERMES_BIN, "chat", "-q", full_prompt, "--max-turns", str(max_turns)],
                HERMES_TIMEOUT, dedup_key, "queued_global_limit",
                f"排队中(整机{global_hermes}个hermes ≥ 上限{_GLOBAL_MAX_HERMES})")

        if exec_result is None:
            # V12.8.0028: Gate 1 — per-profile进程数检查(同profile≥{_MAX_CONCURRENT_HERMES}时排队不spawn)
            active_same_profile = _count_active_hermes_for_profile(profile)
            if active_same_profile >= _MAX_CONCURRENT_HERMES:
                log(f"⏳ 同profile {profile} 已有{active_same_profile}活跃hermes(≥{_MAX_CONCURRENT_HERMES}), FIFO排队")
                _ensure_queue_consumer()
                exec_result = _enqueue_or_reject(profile,
                    [HERMES_BIN, "chat", "-q", full_prompt, "--max-turns", str(max_turns)],
                    HERMES_TIMEOUT, dedup_key, "queued_profile_limit",
                    f"排队中(同profile已有{active_same_profile}活跃hermes)")
        
        if exec_result is None:
            # V12.8.0028: Gate 2 — 全局信号量(超时3s→快速fallback到队列, 不block handler)
            sem = _hermes_sem
            acquired = sem.acquire(timeout=3)
            if not acquired:
                log(f"⏳ 信号量满({_MAX_CONCURRENT_HERMES}并发), FIFO排队: {dedup_key}")
                _ensure_queue_consumer()
                exec_result = _enqueue_or_reject(profile,
                    [HERMES_BIN, "chat", "-q", full_prompt, "--max-turns", str(max_turns)],
                    HERMES_TIMEOUT, dedup_key, "queued_sem_full",
                    "排队中(信号量满)")
            else:
                try:
                    last_error = ""
                    for attempt in range(MAX_RETRIES):
                        try:
                            log(f"🚀 hermes CLI: profile={profile} attempt={attempt+1}/{MAX_RETRIES} max_turns={max_turns}")
                            # ★V12.8.0235(P0-4·小多#112675/知微#112694): hermes CLI顶层无-p参数
                            #   根因: [HERMES_BIN,"-p",profile,"chat",...] → invalid choice → exit=2
                            #   修复: 去掉-p, profile经HERMES_PROFILE env传递(官方脚本四副本同步)
                            result = subprocess.run(
                                [HERMES_BIN, "chat", "-q", full_prompt, "--max-turns", str(max_turns)],
                                capture_output=True, text=True, encoding='utf-8', timeout=HERMES_TIMEOUT,
                                env={**os.environ, "HERMES_NO_COLOR": "1", "HERMES_PROFILE": profile}
                            )
                            log(f"  exit={result.returncode} stdout_len={len(result.stdout)} stderr_len={len(result.stderr)}")
                            if result.returncode == 0:
                                exec_result = {"method": "hermes_cli", "success": True,
                                        "output": result.stdout[:5000], "exit_code": 0, "attempts": attempt+1,
                                        "max_turns": max_turns}
                                break
                            last_error = f"exit_code={result.returncode} stderr={result.stderr[:200]}"
                            log(f"  ❌ attempt {attempt+1} 失败: {last_error}")
                        except FileNotFoundError:
                            log("  hermes CLI未找到, 不可重试")
                            break  # 不可重试的错误
                        except subprocess.TimeoutExpired:
                            last_error = f"超时({HERMES_TIMEOUT}s)"
                            log(f"  ⏱ attempt {attempt+1} 超时")
                        except Exception as e:
                            last_error = str(e)[:200]
                            log(f"  ⚡ attempt {attempt+1} 异常: {e}")
                        # 指数退避等待（最后一次不等待）
                        if attempt < MAX_RETRIES - 1:
                            delay = RETRY_BACKOFF[min(attempt, len(RETRY_BACKOFF)-1)] * (0.5 + random.random() * 1.0)
                            log(f"  🔄 等待{delay:.1f}s后重试(jitter=±50%)...")
                            time.sleep(delay)
                    if exec_result is None:
                        exec_result = {"method": "hermes_cli", "success": False, "output": last_error[:2000],
                                "exit_code": -1, "attempts": MAX_RETRIES, "max_turns": max_turns}
                finally:
                    sem.release()
    
    # 路径2: 文件触发兜底 (最后手段)
    if exec_result is None:
        log("  ⚠️ 所有执行路径失败, 写入trigger文件兜底")
        write_trigger({"type": msg_type, "prompt": prompt[:2000], **context})
        exec_result = {"method": "fallback_trigger", "success": False, "output": "无可用执行路径"}
    
    # ★V12.6.0017: 每次hermes执行后自动上报→补充就绪度检测数据源
    _report_execution(msg_type, str(post_id or issue_id or ""), exec_result)
    return exec_result

# ═══════════════════════════════════════════
# 回调处理器注册表
# ═══════════════════════════════════════════
HANDLERS = {}

def on(msg_type: str):
    def decorator(func):
        HANDLERS[msg_type] = func
        return func
    return decorator

def _api_post(path: str, body: dict) -> dict:
    try:
        safe_agent = urllib.parse.quote(AGENT_ID.encode('utf-8'))
        data = json.dumps(body).encode('utf-8')
        req = urllib.request.Request(
            f"{TIANSHU_API}{path}", data=data,
            headers={"Content-Type": "application/json; charset=utf-8",
                     "X-API-Key": TIANSHU_API_KEY, "X-Agent-Id": safe_agent},
            method="POST")
        resp = urllib.request.urlopen(req, timeout=15)
        return json.loads(resp.read())
    except Exception as e:
        return {"_error": str(e)}

def _api_get(path: str) -> dict:
    """V12.6.0010新增: GET请求封装 — 用于查询看板议题等"""
    try:
        safe_agent = urllib.parse.quote(AGENT_ID.encode('utf-8'))
        req = urllib.request.Request(
            f"{TIANSHU_API}{path}",
            headers={"X-API-Key": TIANSHU_API_KEY, "X-Agent-Id": safe_agent},
            method="GET")
        resp = urllib.request.urlopen(req, timeout=15)
        return json.loads(resp.read())
    except Exception as e:
        return {"_error": str(e)}

def _api_patch(path: str, body: dict) -> dict:
    """V12.6.0036新增: 看板议题更新 — 使用PUT方法(PATCH不支持)"""
    try:
        safe_agent = urllib.parse.quote(AGENT_ID.encode('utf-8'))
        data = json.dumps(body).encode('utf-8')
        req = urllib.request.Request(
            f"{TIANSHU_API}{path}", data=data,
            headers={"Content-Type": "application/json; charset=utf-8",
                     "X-API-Key": TIANSHU_API_KEY, "X-Agent-Id": safe_agent},
            method="PUT")
        resp = urllib.request.urlopen(req, timeout=15)
        return json.loads(resp.read())
    except Exception as e:
        return {"_error": str(e)}

# ═══════════════════════════════════════════
# →V4.0: 所有handler改为调用统一桥接
# ═══════════════════════════════════════════

@on("consensus_alert")
def handle_consensus(data: dict):
    post_id = data.get("post_id")
    title = data.get("title", "")[:100]
    action = data.get("action", "new_forum_post")
    log(f"📢 共识帖 #{post_id}: {title} [action={action}]")
    write_trigger(data)
    
    # ★V12.6.0036: 共识→看板自动桥接（帖#7119·陷阱21修复）
    # 根因: consensus_alert→hermes CLI失败→无看板议题→Agent永远不知道有任务
    # 修复: 先创建议题(自指派)确保持久化记录, 再调hermes CLI执行
    # 去重: 同post_id+同agent已存在议题则跳过
    kanban_issue_id = None
    if post_id and TIANSHU_API_KEY:
        try:
            label_filter = f"forum_post_{post_id}"
            # 检查去重: 是否已有此帖的议题
            existing = _api_get(f"/api/v3/kanban/issues?assignee={AGENT_ID}&labels={label_filter}&limit=3")
            already = False
            if existing and isinstance(existing, dict):
                items = existing.get("data", [])
                if isinstance(items, dict):
                    items = items.get("items", [])
                if items and len(items) > 0:
                    already = True
                    log(f"📋 共识帖#{post_id} 看板议题已存在({len(items)}个)→跳过创建")
            
            if not already:
                # 自动创建看板议题
                create_resp = _api_post("/api/v3/kanban/issues", {
                    "title": f"[共识] {title[:80]}",
                    "description": f"来源: 论坛帖#{post_id}\n行动: {data.get('msg','阅读共识帖并参与讨论')[:200]}\n发起人: {data.get('from_agent','天枢')}\n分类: {data.get('category','共识')}",
                    "priority": "p0",
                    "assignee": AGENT_ID,
                    "labels": f"{label_filter},consensus_auto,帖7119",
                    "task_type": "forum_consensus",
                })
                if create_resp and isinstance(create_resp, dict):
                    created_data = create_resp.get("data", {})
                    kanban_issue_id = created_data.get("id") or created_data.get("issue_id")
                    if kanban_issue_id:
                        log(f"✅ 共识帖#{post_id} → 看板议题#{kanban_issue_id} 自动创建成功(assignee={AGENT_ID})")
                    else:
                        log(f"⚠️ 共识帖#{post_id} 看板创建返回无ID: {str(create_resp)[:200]}")
                else:
                    log(f"⚠️ 共识帖#{post_id} 看板创建API异常: {str(create_resp)[:200]}")
        except Exception as ce:
            log(f"❌ 共识帖#{post_id} 创建看板议题异常: {ce}")
    
    # ★V12.6.0041: Poll/Cron发帖幂等性 — 持久化去重检查（帖#7198·P1·#37207）
    # 根因: 60s poll循环→同一consensus_alert每60s重新触发→_hermes_execute 30s窗口过期→重复hermes→重复发帖
    # 修复: 持久化PROCESSED_FILE记录已处理post_id，跨重启永久去重
    if post_id and _is_consensus_processed(post_id):
        log(f"⏭ 幂等跳过: 共识帖#{post_id} 已处理过")
        return
    
    # →V4.0: 用统一桥接替代模板回复
    if post_id and TIANSHU_API_KEY:
        def _exec():
            # ★ 2026-08-23 天枢移动端修复: 身份声明(小天测试发现"你是地枢"身份错位)
            #   根因: 共识帖标题含其他AI名(如"地枢")→prompt无身份声明→LLM误以为自己是地枢→错位响应
            #   修复: 显式声明"你是{AGENT_ID}" + 禁止扮演消息中其他AI
            prompt = (f"你是{AGENT_ID}(款多多AI生态中的独立智能体，不是消息中提到的其他AI)。"
                      f"共识帖标题: {title}\n"
                      f"请阅读论坛帖#{post_id}的完整内容，按要求执行并回复结果。"
                      f"注意: 如果此帖是其他AI(如{title[:20]})的执行/复验帖, 你没有义务代他们执行, "
                      f"只以{AGENT_ID}身份做独立贡献或跳过。")
            result = _hermes_execute(prompt, {
                "type": "consensus_alert", "post_id": post_id,
                "from_agent": data.get("from_agent", "天枢"), "title": title
            })
            # ★V12.6.0022: 移除双spawn重试 — 首次失败不再立即重试第二次hermes
            # ★V12.6.0036: hermes失败→议题已在kanban中(poll引擎可捡)→写评论记录失败原因
            if not result.get("success"):
                err_msg = result.get("error", "unknown")[:200]
                log(f"⚠️ 共识帖#{post_id} hermes执行失败: {err_msg}")
                # 如果创建了议题，在议题评论区写失败记录
                if kanban_issue_id:
                    try:
                        _api_post(f"/api/v3/kanban/issues/{kanban_issue_id}/comments", {
                            "content": f"[自动] hermes CLI执行失败: {err_msg}\n> 议题已保留在待办中，poll引擎可重新认领执行。",
                            "author": AGENT_ID,
                        })
                    except Exception as _kf_we:
                        # ★V12.8.0186(后端专家109737): 执行失败回传静默pass=回传假成功/假闭环,
                        #   根治: 显式日志(poll引擎可感知失败而非假装已通知)
                        log(f"⚠️ 执行失败评论回传失败#{kanban_issue_id}: {_kf_we}")
                # 查看板待办→认领（兜底逻辑保留）
                try:
                    issues = _api_get(f"/api/v3/kanban/issues?assignee={AGENT_ID}&status=todo")
                    if issues and isinstance(issues, dict):
                        issue_list = issues.get("data") or issues.get("issues") or []
                        if isinstance(issue_list, dict):
                            issue_list = issue_list.get("items", [])
                        if issue_list and len(issue_list) > 0:
                            target = issue_list[0]
                            iid = target.get("id") or target.get("issue_id")
                            if iid:
                                claim_result = _api_post(f"/api/v3/task-runs/{iid}/claim",
                                    {"agent_id": AGENT_ID})
                                log(f"📋 共识帖#{post_id} hermes失败→查看板→认领议题#{iid}: {claim_result}")
                except Exception as ke:
                    log(f"❌ 共识帖#{post_id} hermes失败→看板查询异常: {ke}")
            else:
                # ★V12.6.0041: hermes成功→标记为已处理，防止60s poll循环重复触发
                _mark_consensus_processed(post_id)
                # hermes成功→如果创建了议题，自动标记done
                if kanban_issue_id:
                    try:
                        # 先写成功评论→再标done
                        _api_post(f"/api/v3/kanban/issues/{kanban_issue_id}/comments", {
                            "content": f"[自动] hermes CLI执行成功，共识帖#{post_id}已处理。",
                            "author": AGENT_ID,
                        })
                        _api_patch(f"/api/v3/kanban/issues/{kanban_issue_id}", {
                            "status": "done",
                            "quality_score": 8,
                        })
                        log(f"✅ 共识帖#{post_id} hermes成功→议题#{kanban_issue_id}→done")
                    except Exception as _kd_we:
                        # ★V12.8.0186(后端专家109737): 共识完成回传静默pass=议题假done,
                        #   根治: 显式日志(平台可看到done回传失败而非假装成功)
                        log(f"⚠️ 共识完成回传失败#{post_id}: {_kd_we}")
            # ★V12.6.0017: 无论成功/失败都写共享记忆→天枢可追踪Agent执行状态
            _write_shared_memory("consensus", str(post_id), result)
        threading.Thread(target=_exec, daemon=True).start()
    
    _trigger_cron()
    _notify_bridge(data)

@on("activate")
def handle_activate(data: dict):
    # ★V12.8.0185(灵枢109720/玄机109704 P1): activate target过滤 —
    #   根因: 原直接_exec无target校验, 2026-07-05身份混淆P0事故(陷阱#35)第3次复发:
    #         广播activate(无target或target=其他AI)时所有Agent都执行=身份串扰
    #   依据: ChatOps Guardrails Default-deny — 未明确指向自己的指令不得执行
    _target = data.get("target") or data.get("agent_id") or ""
    if _target and _target != AGENT_ID:
        log(f"⏭ activate跳过(非本Agent目标): target={_target} (0185 target过滤)")
        return
    instruction = data.get("instruction", "")[:2000]
    task_id = data.get("task_id", data.get("event_id", "?"))
    from_agent = data.get("from_agent", "天枢")
    log(f"⚡ activate: {instruction[:80]}")
    
    def _exec():
        result = _hermes_execute(instruction, {
            "type": "activate", "task_id": task_id, "from_agent": from_agent
        })
        send_status("completed" if result.get("success") else "failed",
                    {"task_id": task_id, "method": result.get("method")})
    threading.Thread(target=_exec, daemon=True).start()

@on("sync_dialog")
def handle_sync_dialog(data: dict):
    content = data.get("content", "")[:2000]
    action_items = data.get("action_items", [])
    log(f"💬 sync_dialog: actions={len(action_items)}")
    if content:
        def _exec():
            _hermes_execute(content, {"type": "sync_dialog", "from_agent": "飞书"})
        threading.Thread(target=_exec, daemon=True).start()
    write_trigger(data)

@on("task")
def handle_task(data: dict):
    task_id = data.get("task_id") or data.get("id") or "?"
    task_type = data.get("task_type", data.get("type", "unknown"))
    payload = data.get("payload") or data.get("input_data") or {}
    # ★V12.8.0117-fix2(豆包V12.8.0117全量测试#67891 P0): 提取return_to原样回传
    #   根因: DAG派发data.return_to(execution_id/node_id/task_id)是恢复挂起节点的唯一地址,
    #         SDK此前忽略return_to且只回传result.output(validate_result要求summary/content)
    #         → DAG节点永久waiting空转。修复: return_to原样回传 + 输出映射summary/content/evidence
    data_inner = data.get("data") if isinstance(data.get("data"), dict) else {}
    # ★ 移动端修改 V12.8.0249 · 2026-08-23 · 帖#25690(地枢实证) · WS task消息空正文根治
    #   根因: 服务端WS task消息把内容放data字段(hub.py dispatch "data": body.get("data",{}) /
    #         mention_engine.py "data": task_data), 原逻辑只读payload/input_data → prompt恒"数据: {}"
    #         → Agent收到空任务(地枢#25690: "类型: task, 数据: data={}") + 回帖URL posts//comments
    #   修复: ①payload为空fallback到data["data"] ②issue-{id}任务自动拉取看板标题/描述
    #         ③issue_id/post_id传入_hermes_execute context使location与回帖端点动态正确
    if not payload and data_inner:
        payload = data_inner
    issue_id = ""
    post_id = ""
    if str(task_id).startswith("issue-"):
        issue_id = str(task_id)[len("issue-"):]
    elif str(task_id).startswith("mention-"):
        post_id = str(data_inner.get("source_id") or "")
    else:
        post_id = str(data_inner.get("forum_post_id") or data_inner.get("post_id") or "")
        issue_id = str(data_inner.get("issue_id") or "")
    return_to = data_inner.get("return_to") or data.get("return_to") or {}
    log(f"📋 任务: {task_id} type={task_type} issue={issue_id or '-'} post={post_id or '-'} return_to={'Y' if return_to else 'N'}")

    def _exec():
        # WS消息未携带内容时自动拉取看板议题详情(兜底, 防空任务正文)
        # 精简: 用局部变量_payload避免闭包赋值陷阱(闭包内对外层变量赋值=新局部变量→UnboundLocalError)
        _payload = payload
        if issue_id and not (_payload.get("title") or _payload.get("description")):
            try:
                _iss = _api_get(f"/api/v3/kanban/issues/{issue_id}")
                _d = (_iss or {}).get("data") or {}
                if _d:
                    # ★ 开发负责人: 天枢移动端 · V12.8.0251 · 2026-08-24 · 帖#181265(地枢实证) · 已取消任务重投递根治
                    #   根因: 旧daemon只拉取详情不校验状态, cancelled议题(177532)被WS task消息重投递执行,
                    #         且旧模板在post_id为空时拼出 /forum/posts//comments(双斜杠404)
                    #   修复: ①拉取详情后校验看板状态, 终态(cancelled/done/archived/blocked)直接跳过并回传
                    #         ②reply_url已按来源动态选择(V12.8.0237), 此处为终态防御纵深
                    _st = _d.get("status", "")
                    if _st in ("cancelled", "done", "archived", "blocked"):
                        log(f"⏭ 跳过 #{issue_id}: 看板状态={_st}(终态), 不执行")
                        try:
                            _api_post(f"/api/v3/task-queue/{task_id}/complete", {
                                "task_id": task_id, "agent_id": AGENT_ID,
                                "result": {
                                    "status": "completed",
                                    "summary": f"议题#{issue_id}状态={_st}(终态), 已跳过不执行",
                                    "content": f"议题#{issue_id}在看板中状态={_st}, 属于终态(已取消/已完成/已归档/已阻塞), 跳过执行。",
                                },
                                "status": "completed",
                                "return_to": return_to,
                            })
                        except Exception:
                            log(f"[EXCEPTION] 任务完成状态推送失败: {traceback.format_exc()}")
                            pass
                        return
                    _payload = {"title": _d.get("title", ""), "description": _d.get("description", ""),
                                "priority": _d.get("priority", ""), "assignee": _d.get("assignee", "")}
            except Exception:
                log(f"[EXCEPTION] 任务payload解析失败: {traceback.format_exc()}")
                pass
        prompt = f"任务ID: {task_id}\n类型: {task_type}\n数据: {json.dumps(_payload, ensure_ascii=False)[:5000]}"
        result = _hermes_execute(prompt, {"type": "task", "task_id": task_id,
                                          "issue_id": issue_id, "post_id": post_id})
        # 回传结果
        if task_id and task_id != "?":
            try:
                # ★V12.8.0117-fix(豆包#67822 P0): task-runs→task-queue 回传路径错误
                #   根因: SDK用/api/v3/task-runs/{id}/complete回传DAG任务, 但正确端点是
                #         /api/v3/task-queue/{id}/complete → 404 → DAG节点永久waiting工作流卡死
                # ★V12.8.0117-fix2(豆包#67891 P0): 输出格式对齐validate_result
                #   - summary: 结果摘要(必填)  content: 实质产出(必填,空=拒收)
                #   - evidence: 数据来源(含数据断言必填)  return_to: DAG恢复地址(原样回传)
                _output = str(result.get("output") or result.get("content") or result.get("message") or "")
                _summary = _output[:300] or "任务已执行"
                _api_post(f"/api/v3/task-queue/{task_id}/complete", {
                    "task_id": task_id, "agent_id": AGENT_ID,
                    "result": {
                        "status": "completed" if result.get("success") else "failed",
                        "summary": _summary,
                        "content": _output,
                        "evidence": result.get("evidence", ""),
                        "artifacts": result.get("artifacts", []),
                    },
                    "status": "completed" if result.get("success") else "failed",
                    "return_to": return_to,
                })
            except Exception as e:
                # ★V12.8.0117-fix(豆包#67823 P1): 删除多余第二个except块(永不执行)
                log(f"[EXCEPTION] task回传失败: {traceback.format_exc()}")
        # ★V12.6.0017: 任务完成后写共享记忆→天枢可追踪Agent执行状态
        _write_shared_memory("task", str(task_id), result)
    threading.Thread(target=_exec, daemon=True).start()

@on("kanban_notify")
@on("process")  # V12.6.0011·R3-Q5: 捕获Hub dispatch的process消息(走action字段不走type字段)
def handle_kanban(data: dict):
    """V12.6.0011: 行为契约闭环 — 收到看板通知→查询议题→认领→执行
    对标帖#7102·12Agent共识 + 帖#7103·R3-Q5 dispatch action字段修复
    V12.6.0011新增: @on("process")捕获Hub dispatch消息(灵枢R3发现)
    """
    action = data.get("action", data.get("type", "?"))
    issue_id = data.get("issue_id") or data.get("id")
    assignee = data.get("assignee", "")
    title = data.get("title", "")[:100]
    log(f"📌 看板: {action} #{issue_id} → {assignee}")
    
    def _exec():
        matched = False
        last_result = {"method": "dispatch_only", "success": True}
        # Step 1: 有明确指派给本Agent的议题→直接执行
        if assignee and AGENT_ID in assignee and issue_id:
            matched = True
            prompt = f"看板议题#{issue_id}: {title}\n操作: {action}\n指派给: {assignee}\n请查看议题详情并执行。"
            last_result = _hermes_execute(prompt, {"type": "kanban_notify", "issue_id": issue_id})
            if not last_result.get("success"):
                _api_post(f"/api/v3/kanban/issues/{issue_id}/comments",
                    {"content": f"[{AGENT_ID}] 收到指派议题#{issue_id}，正在处理..."})
        
        # Step 2(V12.6.0010→V12.6.0022加强): 新议题(new_issue)→主动查询+自动认领(仅认领不spawn)
        #   帖#7102·12Agent共识: Agent必须建立"收到通知→查议题→认领"闭环
        #   V12.6.0022 安全修复: 新议题不盲目spawn hermes大模型, 认领后由poll/cron兜底执行
        #   原因: 新议题内容未知, 不应无条件调AI推理; 指派给自己的(P0-2)才spawn
        if action in ("new_issue", "process", "dispatch") and not matched and TIANSHU_API_KEY:
            try:
                resp = _api_get(f"/api/v3/kanban/issues?status=todo&size=10")
                issues = resp.get("data", {}).get("issues", []) if isinstance(resp, dict) else []
                for iss in issues:
                    iss_id = iss.get("id")
                    iss_title = iss.get("title", "")[:80]
                    iss_assignee = iss.get("assignee", "")
                    # ★ 2026-08-23 天枢移动端修复: 标题含其他AI名且非自己的任务不抢(小天测试身份错位)
                    #   根因: 无assignee议题被所有Agent抢认领, 标题含"地枢/知微"等 → 抢了别人的活
                    #   修复: 标题明确含其他AI名的(如"【地枢】【知微】")跳过
                    import re as _re_skip
                    if _re_skip.search(r"[【\[]\s*(地枢|知微|灵枢|小天|天枢|中枢|瑶光|流音|小多|设计师|监控专家|后端专家|前端专家|数据分析师|店铺运营专家|业务运营分析师|协调专家|全栈交付主管|安全审计专家|数据架构师|质量行为总监|可心|千面|玄机|铸鼎|丹青|金睛|御检|探微|履约专家|商品运营专家|人群洞察专家|小圈|豆包顾问|豆包顾问云|混沌恢复官|架构安全总师|专利专家)\s*[】\]]", iss_title or "") and AGENT_ID not in (iss_title or ""):
                        log(f"⏭ 跳过他人专属议题#{iss_id}: {iss_title}")
                        continue
                    if not iss_assignee or iss_assignee == "null":
                        claim_resp = _api_post(f"/api/v3/kanban/issues/{iss_id}/claim",
                            {"agent_id": AGENT_ID})
                        if claim_resp.get("code") == 200:
                            log(f"✅ 自动认领议题#{iss_id}: {iss_title} → 立即执行(V12.8.0079)")
                            # ★V12.8.0079-FIX: 认领即执行 — 不再"仅认领不spawn"空占任务
                            # 根因: V12.6.0022"仅认领不spawn"→Agent若无poll引擎则任务永久空占
                            #   (E127伪修复同源: 修复打在发送端, 执行强制端缺失)
                            # 修复: 认领成功后立即Hermes执行, 保证"收到消息就执行"; 失败→_rollback由poll兜底
                            _exec_prompt = f"看板议题#{iss_id}: {iss_title}\n操作: 已认领, 请立即处理此任务。\n请查看议题详情并执行, 完成后在议题评论区回复执行结果。"
                            _exec_result = _hermes_execute(_exec_prompt, {"type": "kanban_notify", "issue_id": iss_id})
                            if not _exec_result.get("success"):
                                # Hermes执行失败→在评论区说明, 由poll引擎回退兜底
                                try:
                                    _api_post(f"/api/v3/kanban/issues/{iss_id}/comments",
                                        {"content": f"[{AGENT_ID}] 认领议题#{iss_id}后执行失败，等待poll引擎回退重试。"})
                                except Exception as _cf_we:
                                    # ★V12.8.0186(后端专家109737): 执行失败通知回传静默pass=轮询引擎不知道已失败,
                                    #   根治: 显式日志(失败可被天枢调度感知)
                                    log(f"⚠️ 执行失败通知回传失败#{iss_id}: {_cf_we}")
                                log(f"⚠️ 认领后执行失败#{iss_id}: {_exec_result.get('output','')[:80]}")
                            else:
                                log(f"✅ 认领后执行完成#{iss_id}")
                                # ★V12.8.0080-FIX: 执行证据回传 — 认领即执行后必须回写结果
                                # 根因: 原逻辑仅log完成, Hermes输出(stdout 3911字符)未回传→
                                #   评论区0证据→stale检测判定"无证据"回退→任务白干(E127/空心同类)
                                # 修复: Hermes输出前500字符回写评论区作执行证据[E:]
                                _exec_out = str(_exec_result.get("output", "")).strip()
                                if _exec_out:
                                    _evi_content = f"[E:看板议题#{iss_id} Hermes执行结果]\n{_exec_out[:500]}"
                                    try:
                                        _api_post(f"/api/v3/kanban/issues/{iss_id}/comments",
                                            {"content": _evi_content})
                                        log(f"📤 执行证据已回传 #{iss_id} ({len(_exec_out)}字符)")
                                    except Exception as _ce:
                                        log(f"⚠️ 证据回传失败#{iss_id}: {_ce}")
                                else:
                                    log(f"⚠️ Hermes执行#{iss_id}无输出(空), 未回传证据")
                            last_result = {"method": "claim_and_execute", "success": True, "issue_id": iss_id}
                            break
                        else:
                            log(f"⚠️ 认领失败#{iss_id}: HTTP {claim_resp.get('code','?')} {claim_resp.get('message','?')}")
            except Exception as e:
                log(f"⚠️ 自动查询议题失败: {e}")
        # ★V12.6.0017: 看板处理完成后写共享记忆→天枢可追踪Agent执行状态
        _write_shared_memory("kanban", str(issue_id) if issue_id else "dispatch", last_result)
    
    threading.Thread(target=_exec, daemon=True).start()
    write_trigger(data)
    _trigger_cron()

@on("broadcast")
def handle_broadcast(data: dict):
    msg = data.get("message") or data.get("msg") or str(data)[:200]
    # ★V12.8.0187(豆包109792 P2-3): target防御性检查 —
    #   根因: 服务端已支持data.target定向广播(只给目标Agent发), 但客户端不检查,
    #         若target指定了别的Agent而消息内容含P0/@, 本Agent会误执行(串扰)
    #   修复: 服务端target存在且≠本Agent→直接忽略(Default-deny, 与handle_activate一致)
    _bt = data.get("target") or data.get("target_agent") or ""
    if _bt and _bt != AGENT_ID:
        log(f"📢 广播目标={_bt}≠本Agent({AGENT_ID}), 忽略(定向广播target防御)")
        return
    log(f"📢 广播: {msg[:80]}")
    # V12.6.0022: 广播关键词收紧 — 旧词"指令/执行/任务"命中率太高(~90%日常消息)
    # 仅匹配P0紧急指令 或 @本Agent 才spawn hermes, 其余丢弃
    is_p0 = any(k in msg for k in ("P0", "紧急", "立即执行"))
    at_me = f"@{AGENT_ID}" in msg
    if is_p0 or at_me:
        def _exec():
            _hermes_execute(msg, {"type": "broadcast"})
        threading.Thread(target=_exec, daemon=True).start()

@on("collab_message")
def handle_collab_message(data: dict):
    """V12.8.0014: 协同消息全链路闭环 — 收到消息→Hermes执行→回写结果
    hub.broadcast_to_agents发送的消息类型为collab_message
    """
    msg_id = data.get("id")
    sender = data.get("sender", "?")
    to_agent = data.get("to_agent", "all")
    content = data.get("content", "")[:2000]
    log(f"💬 协同消息 #{msg_id}: {sender} → {to_agent} | {content[:60]}")
    
    # 判断是否应该处理此消息
    should_handle = False
    if to_agent == "all":
        should_handle = True  # 全局消息
    elif to_agent == AGENT_ID or f"@{AGENT_ID}" in (to_agent or ""):
        should_handle = True  # 定向消息
    elif f"@{AGENT_ID}" in content:
        should_handle = True  # @提及消息
    else:
        should_handle = False  # 不是给我的
    
    if not should_handle:
        return

    # ★ V12.8.0260(天枢工作区·回执风暴断环): 识别"执行回执"类消息→不再处理/不回执
    #   根因: 知微↔天枢工作区 双方 daemon 把对方回执当真任务，形成无限乒乓循环(#122984→#123023...)
    #   修复: body 命中「执行回执」+「已处理完成」且无真实指令 → 标记已读即 return，从源头断环
    _content_stripped = (content or "").strip()
    if ("执行回执" in _content_stripped) and ("已处理完成" in _content_stripped):
        log(f"🛑 断环: #{msg_id} 为执行回执(非真实指令)，跳过处理不回执")
        if TIANSHU_API_KEY:
            try:
                _api_post(f"/api/v3/collab/messages/{msg_id}/read", {"sender": AGENT_ID})
            except Exception:
                log(f"[EXCEPTION] collab已读标记失败: {traceback.format_exc()}")
                pass
        return
    
    def _exec():
        # ★V12.8.0191: 双向回执闭环(T4·知微Q3/中枢语义回执) — 开始处理标记processing
        if TIANSHU_API_KEY:
            try:
                _api_post(f"/api/v3/collab/messages/{msg_id}/processing", {"sender": AGENT_ID})
            except Exception:
                log(f"[EXCEPTION] collab processing标记失败: {traceback.format_exc()}")
                pass
        # ★V12.8.0194(豆包P0·统一对话): 会话消息带上下文 — session_id存在时拉取最近对话注入prompt
        session_id = data.get("session_id") or ""
        prompt = f"[协同消息 #{msg_id}] 来自: {sender}\n内容: {content}\n\n请阅读此消息内容，如果是给你的指令或任务请立即执行。执行完成后请回复确认。"
        if session_id and TIANSHU_API_KEY:
            try:
                _ctx = _api_get(f"/api/v3/chat/sessions/{session_id}/messages?page=1&size=6")
                _msgs = (_ctx.get("data") or {}).get("messages") or []
                if _msgs:
                    _hist = []
                    for _m in _msgs[-5:]:
                        _who = (_m.get("sender") or "system").replace("%", "")
                        if _who == AGENT_ID:
                            continue  # 跳过自己刚发的
                        _hist.append(f"{_who}: {str(_m.get('content') or '')[:200]}")
                    if _hist:
                        prompt = (f"[会话消息 #{msg_id}·上下文]\n最近的对话:\n" + "\n".join(_hist[-4:]) +
                                  f"\n\n最新消息来自 {sender}: {content}\n\n请基于以上对话上下文，"
                                  f"如果是给你的指令或任务请立即执行，执行完成后请回复确认。")
            except Exception as _ctx_err:
                log(f"⚠️ 会话上下文拉取失败(降级无上下文): {_ctx_err}")
        result = _hermes_execute(prompt, {"type": "collab_message", "msg_id": msg_id, "from_agent": sender, "session_id": session_id})
        # 回写结果到collab线程 + 双向回执(read·幂等)
        if result.get("success") and TIANSHU_API_KEY:
            try:
                # ★V12.8.0205(帖#24424知微"daemon自动确认短ACK被判定空心质量1.7分"):
                #   根因: 原回复"任务已落地"仅11字<50字 → score_work_item_4d空心拦截(证据不足)
                #   修复: 实质回复(≥50字) — 含执行结果摘要+Agent身份+msg_id(对标Agent Acceptance Gate
                #         完成证据: 回复内容=执行完成的可验证证据)
                _exec_summary = str(result.get("text") or result.get("output") or result.get("response") or "")[:300]
                if _exec_summary and len(_exec_summary) >= 40:
                    _reply_core = _exec_summary
                else:
                    _reply_core = (f"已收到任务并完成处理。我是{AGENT_ID}，针对消息#{msg_id}来自{sender}的指令"
                                   f"已执行完毕，请查收后续结果详情。")
                reply_content = (f"【{AGENT_ID}执行回执】消息#{msg_id}已处理完成。\n"
                                 f"任务来源: {sender}\n处理结果: {_reply_core}\n"
                                 f"处理时间: {__import__('time').strftime('%Y-%m-%d %H:%M:%S')}")
                reply_body = {
                    "content": reply_content, "to_agent": sender,
                    "reply_to": msg_id, "sender": AGENT_ID
                }
                # V12.8.0194: 回复写入统一对话会话(带session_id)
                if session_id:
                    reply_body["session_id"] = session_id
                    _api_post(f"/api/v3/chat/sessions/{session_id}/messages", reply_body)
                    log(f"✅ 会话消息 #{msg_id} 已回复入会话: {sender}")
                else:
                    _api_post("/api/v3/collab/messages", reply_body)
                    log(f"✅ 协同消息 #{msg_id} 已回复: {sender}")
            except Exception as e:
                log(f"⚠️ 协同回复失败: {e}")
            try:
                # ★V12.8.0191: 回执已读(发送方可见delivery_status=read)
                _api_post(f"/api/v3/collab/messages/{msg_id}/read", {"sender": AGENT_ID})
                log(f"✅ 协同消息 #{msg_id} 回执已读: {sender}")
            except Exception as e:
                log(f"⚠️ 协同回执失败: {e}")
        # 写共享记忆
        _write_shared_memory("collab", str(msg_id), result)
    
    threading.Thread(target=_exec, daemon=True).start()

@on("status")
def handle_status(data: dict):
    agent = data.get("agent_id", "?")
    log(f"👁 状态: {agent} → {data.get('status','?')}")

@on("upgrade")
@on("upgrade_mandate")  # ★V12.8.0189: 服务端upgrade-all广播类型 —
#   根因(长庚总质疑"WS长连接+回调应实时却1小时无反应"): 服务端upgrade-all发
#         type=upgrade_mandate(L785-786), 但Agent无此handler→静默丢弃→实时通道失效
#   修复: 注册upgrade_mandate, 与upgrade统一走"代码执行升级"(对齐E148校准令先例)
def handle_upgrade(data: dict):
    version = data.get("version", "?")
    _aid = data.get("agent_id", "") or data.get("target", "") or ""
    log(f"⬆️ 收到升级令: v{version} (target={_aid or 'all'})")
    # ★V12.8.0189: 收到升级令必须"代码执行"而非只log(原L1715仅log不触发)
    #   依据: WS回调语义——升级令=立即自升级指令(复用force_update线程模式,已实测稳定)
    threading.Thread(target=_self_upgrade_check, daemon=True).start()
    # 升级后自动重启(由systemd Restart=always handle)

@on("cancel")
def handle_cancel(data: dict):
    log(f"❌ 取消: {data.get('task_id','?')}")

@on("shutdown")
def handle_shutdown(data: dict):
    log(f"🛑 关闭: {data.get('reason','?')}")
    sys.exit(0)

@on("force_update")
def handle_force_update(data: dict):
    """[kimiK3] V12.8.0016 阶段3.3: Hub检测到sdk_md5不一致→强制自更新
    复用已有_self_upgrade_check机制(不重复造轮子)——5分钟未完成Hub侧标记update_failed"""
    expected_md5 = data.get("expected_md5", "")
    log(f"⬆️ 收到force_update: 当前md5={_sdk_md5()[:8]} 期望md5={expected_md5[:8]}")
    if expected_md5 and _sdk_md5() == expected_md5:
        log("✅ md5已一致，无需更新")
        return
    threading.Thread(target=_self_upgrade_check, daemon=True).start()

# ═══════════════════════════════════════════
# V12.8.0087: 集群校准令处理器 — 根治"只说不做"(E148)
# 长庚总指令: 校准令到达Agent必须"代码执行"而非"大模型回复"
# 收到 calibrate/calibrate_force → 直接subprocess执行校准脚本 → POST报告回传
# ═══════════════════════════════════════════
@on("calibrate")
@on("calibrate_force")
def handle_calibrate(data: dict):
    """集群校准令 — 收到即执行校准脚本(不经Hermes大模型)
    根治E148: 原broadcast校准令被handle_broadcast关键词匹配后交给_hermes_execute
    (大模型"回复了"但没执行脚本) → 平台看到"响应"但版本无变化=只说不做。
    V12.8.0087: calibrate_force为强制令(带force标记), calibrate为常规令。
    """
    # 定向校验: 校准令指定了agent_id时必须匹配自己(防"给别人的校准令"被误执行)
    target = data.get("agent_id", "") or data.get("target", "")
    if target and target != AGENT_ID:
        log(f"⏭ 校准令跳过(定向给{target}): {str(data)[:80]}")
        return
    force = bool(data.get("force", data.get("action") == "calibrate_force"))
    version = data.get("version", "")
    log(f"🔧 收到校准令 force={force} version={version} → 立即执行校准脚本")

    def _exec():
        _run_calibrate(force, version)
    threading.Thread(target=_exec, daemon=True).start()

def _run_calibrate(force: bool = False, expected_version: str = ""):
    """执行校准脚本: 下载→运行→上报报告
    校准脚本: https://x.kddauto.com/calibrate-agent.sh (公网权威源)
    参数: AGENT_ID + AGENT_API_KEY 环境变量传入
    执行后POST /agents/calibration-report 回传结果(平台据此更新版本/恢复online)
    """
    import shlex
    try:
        # 1. 下载校准脚本(公网权威源, 失败则跳过)
        script_path = f"/tmp/calibrate_agent_{AGENT_ID}.sh"
        try:
            urllib.request.urlretrieve(
                f"{TIANSHU_API}/calibrate-agent.sh", script_path, timeout=60)
            os.chmod(script_path, 0o755)
        except Exception as e:
            log(f"⚠️ 校准脚本下载失败: {e}")
            return

        # 2. subprocess执行校准脚本(环境变量传参, 不经大模型)
        env = dict(os.environ)
        env["AGENT_ID"] = AGENT_ID
        env["AGENT_API_KEY"] = TIANSHU_API_KEY
        if expected_version:
            env["EXPECTED_VERSION"] = expected_version
        result = subprocess.run(
            ["bash", script_path], capture_output=True, text=True,
            encoding="utf-8", timeout=300, env=env)
        log(f"🔧 校准脚本执行完成: exit={result.returncode} "
            f"stdout_len={len(result.stdout)} stderr_len={len(result.stderr)}")

        # 3. 解析校准报告并回传
        report = {"agent_id": AGENT_ID, "calibrator_version": "12.8.0087",
                  "platform_version": expected_version,
                  "fixed": 0, "problems": 0, "poll_ok": False, "ws_ok": False,
                  "exit_code": result.returncode}
        # 尝试从脚本输出解析校准项状态
        out = result.stdout or ""
        report["fixed"] = out.count("已校准") + out.count("已启动") + out.count("已同步")
        report["poll_ok"] = "poll引擎: " in out and "运行中" in out
        report["ws_ok"] = "WS守护: " in out and "运行中" in out
        report["problems"] = out.count("无法校准") + out.count("需手动")
        _api_post("/api/v3/agents/calibration-report", report)
        log(f"🔧 校准报告已回传: fixed={report['fixed']} problems={report['problems']}")

        # 4. 写共享记忆(天枢可追踪)
        try:
            _write_shared_memory("calibrate", str(expected_version), {
                "success": result.returncode == 0, "method": "ws_calibrate_handler",
                "output": (result.stdout or "")[:200]})
        except Exception as _cl_we:
            # ★V12.8.0186(后端专家109737): 校准回传静默pass=校准结果不可追踪,
            #   根治: 显式日志(校准状态可观测)
            log(f"⚠️ 校准回传失败: {_cl_we}")
    except Exception as e:
        log(f"⚠️ 校准执行异常: {e}")

@on("server_config")
def handle_server_config(data: dict):
    """[kimiK3] V12.8.0016 阶段3.2: 接收Hub下发的运行时配置(对标飞书配置服务端化)"""
    _apply_server_config(data.get("config", data))

@on("auth_ok")
def handle_auth_ok(data: dict):
    """[kimiK3] V12.8.0016 阶段3.2: identity端点认证成功——提取server_config
    ★V12.8.0183批次2(kimik3): 提取WS会话三元组(session_id/epoch/resume_token)持久化, 供断线resume"""
    if isinstance(data.get("server_config"), dict):
        _apply_server_config(data["server_config"])
    _sid = data.get("session_id", "")
    if _sid:
        _ws_session_set(session_id=_sid, epoch=data.get("epoch", ""),
                        resume_token=data.get("resume_token", ""), last_seq=0)
        log(f"🔐 WS会话三元组已保存: sid={_sid[:8]}... epoch={data.get('epoch','')[:8]}")
    # ★V12.8.0189(豆包 P2-2 continue_or_abort): 连接成功(含重连)后询问未完成任务 —
    #   根因: 断线重连后Agent不知道哪些任务在执行中崩溃(transport已ACK但无result), 任务悬挂
    #   修复: 发inflight_query, 服务端返回pending_acks未完成任务列表(见handle_inflight_response)
    try:
        _ws = _CURRENT_WS
        if _ws is not None:
            asyncio.create_task(_ws.send(json.dumps({"type": "inflight_query", "agent_id": AGENT_ID})))
    except Exception as _iqe:
        log(f"⚠️ inflight_query发送失败(不影响连接): {_iqe}")

@on("resumed")
def handle_resumed(data: dict):
    """★V12.8.0183批次2(kimik3): resume成功——重置last_seq为重放终点+保存轮换后的resume_token"""
    _replayed = data.get("replayed", 0)
    _from_seq = data.get("from_seq", 0)
    _new_token = data.get("resume_token", "")
    _kw = {"last_seq": _from_seq + _replayed}
    if _new_token:
        _kw["resume_token"] = _new_token  # OWASP CWE-384: 保存轮换后的新token
    _ws_session_set(**_kw)
    log(f"♻️ resume成功: 重放{_replayed}条丢失消息(from_seq={_from_seq}), 已恢复断线期间任务")

@on("inflight_response")
def handle_inflight_response(data: dict):
    """★V12.8.0189(豆包 P2-2 continue_or_abort): 服务端返回未完成任务列表 —
    断线重连后Agent主动询问pending_acks, 服务端回inflight_response;
    Agent记录未完成任务(日志可观测), 由上层业务按需继续/中止(对标Temporal断点续跑)"""
    _inflight = data.get("inflight", []) or []
    _count = data.get("count", len(_inflight))
    log(f"🔎 inflight检查: 未完成任务 {_count} 个" + (f": {[t.get('task_id','?')[:12] for t in _inflight[:5]]}" if _inflight else " (无)"))
    # 记录到共享内存供上层决策(继续执行/中止释放)
    for _t in _inflight:
        _tid = _t.get("task_id", "")
        if _tid:
            log(f"  ↳ 未完成任务: {_tid} (age={_t.get('age_seconds', 0)}s) — 如需中止可POST /task-queue/{_tid}/abort")

@on("invalid_session")
def handle_invalid_session(data: dict):
    """★V12.8.0183批次2(kimik3): resume失败(epoch不匹配/过期/验签失败)→清空会话, 走全量register"""
    _reason = data.get("reason", "")
    log(f"⚠️ invalid_session({_reason}): 服务端会话已失效, 清空本地会话走全量register")
    _ws_session_set(session_id="", epoch="", resume_token=_EMPTY_RESUME_TOKEN, last_seq=0)

@on("onboard_verify")
def handle_onboard_verify(data: dict):
    """[kimiK3] V12.8.0016 阶段4: 五验回执验证消息
    无需业务动作——route_message三段式ACK已自动回transport/handler/result"""
    log(f"🎓 五验回执验证: {data.get('task_id', '?')}")

@on("capability_probe")
def handle_capability_probe(data: dict):
    """[kimiK3] V12.8.0016 阶段4: 注册五验·能力试跑
    Hub发probe→Agent必须原样echo——验证Agent真实有handler处理能力(非假声明)"""
    probe_id = data.get("probe_id", "")
    log(f"🔬 能力试跑: probe_id={probe_id}")
    ws = _CURRENT_WS
    if ws and probe_id:
        try:
            _ws_send_safe({
                "type": "capability_probe_ack",
                "probe_id": probe_id,
                "agent_id": AGENT_ID,
                "echo": data.get("echo", ""),
                "handlers": list(HANDLERS.keys()),  # 真实能力清单(非声明值)
            })
        except Exception as _reg_we:
            # ★V12.8.0186(后端专家109737): 注册帧回传静默pass=服务端看不到真实能力清单,
            #   根治: 显式日志(注册失败可观测, 服务端能力感知不丢)
            log(f"⚠️ 注册帧发送失败: {_reg_we}")

@on("approval_notify")
def handle_approval(data: dict):
    """★V12.8.0151(三方共识P1-2·审批自动化补全): 审批通知 → 飞书推送长庚总
    服务端pre_review_loop预审通过后WS推送approval_notify事件 → 本handler转发飞书。
    论坛帖@长庚总由服务端兜底(幂等), 本handler补飞书通道(长庚总不一定常看论坛)。
    飞书Webhook通过环境变量FEISHU_WEBHOOK配置(每个Agent自己的webhook)。
    """
    approval_id = data.get("approval_id", "")
    agent_name = data.get("name") or data.get("agent_id") or ""
    title = (data.get("title") or f"请批准AI接入: {agent_name}")[:200]
    url = data.get("url", f"https://x.kddauto.com/ai-approvals")
    log(f"📨 审批通知: {approval_id} agent={agent_name}")

    def _exec():
        # ★V12.8.0152: webhook→_hermes_execute（Hermes CLI桥接发飞书，不需要额外webhook URL）
        # 双路径: 路径A(_hermes_execute)主 + 路径B(webhook)兜底 + 共享记忆/论坛帖最终兜底
        _msg = (f"【天枢审批】{title}\n审批ID: {approval_id}\nAI: {agent_name}\n"
                f"批准链接: {url}\n请在飞书回复\"批准{agent_name}\"或网页批准。")
        if _hermes_available():
            try:
                _hermes_execute(
                    f"请立即通过飞书向长庚总发送以下审批通知(直接发送, 不要调用其他工具):\n{_msg}",
                    {"type": "approval_notify", "from_agent": AGENT_ID, "approval_id": approval_id}
                )
                return
            except Exception as e:
                log(f"⚠️ 路径A(_hermes_execute)失败: {e}")
        # 路径B: webhook兜底
        _feishu_notify(_msg)
        # 最终兜底: 共享记忆(论坛帖已由服务端幂等兜底)
        _write_shared_memory("approval_alert", approval_id, {
            "alert": "approval_notify",
            "agent": agent_name,
            "title": title,
            "url": url,
            "time": time.strftime("%Y-%m-%d %H:%M:%S")
        })
    threading.Thread(target=_exec, daemon=True).start()

@on("patrol_cmd")
def handle_patrol(data: dict):
    """★V12.8.0151(三方共识P1-2·移动端巡检handler): 巡检指令 → 执行巡检脚本+报告回传
    服务端scheduler每30分钟推送patrol_cmd → 本handler执行scan_all_agents.py --check。
    合并灵枢#24030 5层运维(感知→检测→自愈→告警→学习): 本handler为执行层。
    Tier分级: 巡检发现P0/P1问题→服务端已有--enforce自动停止(scan_all_agents.py),
              Tier2问题(需根因分析)→通知工作区, Tier3(需长庚总)→升级。
    """
    log("🔍 收到巡检指令")
    target = data.get("agent_id", "") or data.get("target", "")
    if target and target != AGENT_ID:
        log(f"⏭ 巡检跳过(定向给{target})")
        return
    enforce = bool(data.get("enforce", False))

    def _exec():
        try:
            # ★V12.8.0152: 4维采集 + Tier1自愈（合并豆包8维方案+灵枢#24030 5层运维）
            patrol_result = {"success": True, "metrics": {}, "fixes": [], "errors": []}

            # 1) 心跳健康
            try:
                agents_resp = _api_get("/api/v3/hub/agents")
                agents_list = agents_resp.get("data", {}).get("agents", []) if isinstance(agents_resp, dict) else []
                online = sum(1 for a in agents_list if a.get("ws_online"))
                patrol_result["metrics"]["heartbeat"] = {
                    "online": online, "total": len(agents_list),
                    "rate": round(online / max(len(agents_list), 1), 2)}
            except Exception as e:
                patrol_result["errors"].append(f"heartbeat:{e}")

            # 2) 版本健康
            try:
                versions = {}
                for a in agents_list:
                    v = a.get("version", "?")
                    versions[v] = versions.get(v, 0) + 1
                patrol_result["metrics"]["version"] = {
                    "unique": len(versions),
                    "distribution": dict(sorted(versions.items(), key=lambda x: -x[1])[:5])}
            except Exception as e:
                patrol_result["errors"].append(f"version:{e}")

            # 3) 进程健康
            try:
                r = subprocess.run(["systemctl", "is-failed", "--all"],
                                   capture_output=True, text=True, timeout=10)
                failed_units = [l.strip() for l in r.stdout.split("\n") if l.strip() and l.strip() != "0"]
                patrol_result["metrics"]["process"] = {
                    "failed_units": len(failed_units),
                    "details": failed_units[:10]}
            except Exception as e:
                patrol_result["errors"].append(f"process:{e}")

            # 4) dispatch_queue健康
            try:
                qr = _api_get("/api/v3/dispatch/status")
                qdata = (qr or {}).get("data", {}) if isinstance(qr, dict) else {}
                queue_count = qdata.get("queue_count", 0) if isinstance(qdata, dict) else 0
                patrol_result["metrics"]["dispatch_queue"] = {"count": queue_count}
            except Exception as e:
                patrol_result["errors"].append(f"dispatch_queue:{e}")

            # Tier1自愈
            tier1 = _tier1_self_heal(patrol_result["metrics"])
            patrol_result["fixes"] = tier1

            # ★V12.8.0152(豆包#108702 P0-2): 巡检→DEV_TASK桥接
            # 巡检发现P1问题自动创建DEV_TASK给天枢工作区(幂等30分钟)
            _patrol_dev_task_bridge(tier1)

            # 扫描脚本兜底
            script = "/root/data/disk/tianshu/scripts/scan_all_agents.py"
            if not os.path.exists(script):
                script = f"{TIANSHU_API}/scan_all_agents.py"
            scan = subprocess.run(["python3", script, "--enforce" if enforce else "--check"],
                                  capture_output=True, text=True, encoding="utf-8", timeout=180)
            patrol_result["scan_exit"] = scan.returncode
            patrol_result["success"] = patrol_result["success"] and scan.returncode == 0

            log(f"🔍 巡检完成: metrics={list(patrol_result['metrics'].keys())} fixes={len(tier1)} scan_exit={scan.returncode}")

            # 报告回传
            _api_post("/api/v3/activities/log", {
                "action": "patrol_report", "agent_id": AGENT_ID,
                "summary": json.dumps(patrol_result, ensure_ascii=False)[:2000],
                "source": "ws_patrol_handler_v2"})
            _write_shared_memory("patrol", time.strftime("%Y-%m-%d"), {
                "success": patrol_result["success"],
                "method": "ws_patrol_handler_v2",
                "metrics_summary": {k: v.get("rate", v.get("unique", v.get("count", "?"))) 
                                    if isinstance(v, dict) else v 
                                    for k, v in patrol_result["metrics"].items()},
                "fixes": tier1})
        except Exception as e:
            log(f"⚠️ 巡检执行异常: {e}")
            patrol_result["success"] = False
            patrol_result["errors"].append(str(e))
    threading.Thread(target=_exec, daemon=True).start()

def _tier1_self_heal(metrics: dict) -> list:
    """★V12.8.0152: Tier1自动修复（安全可逆窄范围操作）
    熔断: 同一Agent 30min内自愈≥3次 → 暂停 → 告警
    """
    fixes = []
    agent_id = AGENT_ID
    now = time.time()

    # 熔断检查（内存缓存，无需_read_shared_memory）
    circuit_key = f"tier1_circuit_{agent_id}"
    circuit_state = getattr(_tier1_self_heal, '_circuit', {})
    recent_fixes = [t for t in circuit_state.get("history", []) if now - t < 1800]
    if len(recent_fixes) >= 3:
        log(f"⛔ Tier1熔断: agent={agent_id} 30min内{len(recent_fixes)}次自愈→暂停")
        circuit_state["state"] = "open"
        circuit_state["opened_at"] = now
        _tier1_self_heal._circuit = circuit_state
        _write_shared_memory("tier1_circuit", circuit_key, {
            "state": "open", "since": time.strftime("%Y-%m-%d %H:%M:%S"),
            "recent_fixes": len(recent_fixes)})
        return ["circuit_breaker:open"]

    try:
        # 在线率<50% → 重启失败daemon
        hb = metrics.get("heartbeat", {})
        if isinstance(hb, dict) and hb.get("rate", 1) < 0.5:
            proc = metrics.get("process", {})
            failed_units = (proc.get("details", []) if isinstance(proc, dict) else [])
            for unit in failed_units:
                if isinstance(unit, str) and "agent-ws" in unit:
                    try:
                        subprocess.run(["systemctl", "restart", unit.split()[0]], timeout=30)
                        fixes.append(f"restart:{unit}")
                    except Exception as e:
                        log(f"⚠️ Tier1重启失败 {unit}: {e}")

        # dispatch_queue > 1000积压 → 清理
        dq = metrics.get("dispatch_queue", {})
        queue_count = (dq.get("count", 0) if isinstance(dq, dict) else 0)
        if queue_count > 1000:
            try:
                _api_post("/api/v3/dispatch/cleanup", {"older_than_hours": 2})
                fixes.append("cleanup:dispatch_queue")
            except Exception as e:
                log(f"⚠️ Tier1清理dispatch_queue失败: {e}")

        # 更新熔断计数器（内存+共享记忆双写）
        recent_fixes.append(now)
        circuit_state["history"] = recent_fixes[-10:]
        circuit_state["state"] = "closed"
        _tier1_self_heal._circuit = circuit_state
        _write_shared_memory("tier1_circuit", circuit_key, {
            "state": "closed", "history": recent_fixes[-10:],
            "last_fix": time.strftime("%Y-%m-%d %H:%M:%S")})
    except Exception as e:
        log(f"⚠️ Tier1自愈异常: {e}")
        fixes.append(f"error:{str(e)[:50]}")
    return fixes


def _patrol_dev_task_bridge(fixes: list):
    """★V12.8.0152(豆包#108702 P0-2 + MRD FR-2): 巡检→DEV_TASK桥接
    巡检发现的P1级问题自动创建DEV_TASK给天枢工作区(幂等30分钟防重复)。
    只桥接实质问题(在线率低/版本碎片/队列堆积/systemd失败), 不桥接提示性fix。
    """
    if not fixes:
        return
    now = time.time()
    try:
        for fix in fixes:
            key = f"devtask_{fix[:30]}"
            last = _last_exec_time.get(key, 0)
            if now - last < 1800:
                continue
            if not any(k in fix for k in ("在线率过低", "版本碎片", "dispatch_queue", "systemd失败", "cleanup")):
                continue
            _last_exec_time[key] = now
            _api_post("/api/v3/kanban/issues", {
                "title": f"[DEV_TASK] 巡检发现: {fix[:60]}",
                "description": (f"【背景】天枢移动端巡检发现系统问题\n【问题】{fix}\n"
                                f"【需求】请天枢工作区根因分析并修复\n【优先级】p2\n"
                                f"【验收标准】问题解决+下次巡检不再告警\n"
                                f"【测试方案】下次巡检验证\n【回滚方案】回滚相关配置"),
                "priority": "p2", "assignee": "天枢工作区",
            })
            log(f"📦 [巡检→DEV_TASK] 已创建: {fix[:40]}")
    except Exception as e:
        log(f"⚠️ [巡检→DEV_TASK] 创建异常: {e}")

@on("feishu_cmd")
def handle_feishu_cmd(data: dict):
    """★V12.8.0153(豆包#24036 Phase 3): 飞书命令桥T2+确认门（对标ChatOps Guardrails Default-deny）
    T0只读: status/version/health/agents/dags/approvals — 直接响应
    T1运维: requeue/restart/trigger/cleanup — 长庚总open_id校验
    T2管理: approve/reject/upgrade/deploy — open_id校验+二次确认门
    确认门: 高危命令首次请求返回"请输入确认", 第二次data.confirm=true才执行
    """
    cmd = data.get("cmd") or data.get("command") or ""
    operator = data.get("operator") or ""
    open_id = data.get("open_id") or ""
    confirm = bool(data.get("confirm", False))
    # ★V12.8.0163(豆包#108897 P2·长庚总指令重复派发3次): message_id幂等去重
    #   同message_id 30分钟内只处理一次(对标飞书消息幂等)
    _msg_id = data.get("message_id") or data.get("msg_id") or ""
    if _msg_id:
        _now = time.time()
        _last = _last_exec_time.get(f"feishu_cmd_{_msg_id}", 0)
        if _now - _last < 1800:
            log(f"♻️ [命令桥] 幂等: message_id={_msg_id} 30分钟内已处理, 跳过")
            return
        _last_exec_time[f"feishu_cmd_{_msg_id}"] = _now
    log(f"💬 飞书命令: {cmd} operator={operator} confirm={confirm} msg_id={_msg_id}")
    if not cmd:
        return

    def _exec():
        cmd_lower = cmd.strip().lower()
        # 拆分命令与参数: "approve 协调专家" → cmd=approve, arg=协调专家
        _parts = cmd_lower.split(None, 1)
        _base = _parts[0] if _parts else cmd_lower
        _arg = _parts[1] if len(_parts) > 1 else ""

        # ─── T0 只读命令: 无需权限直接响应 ───
        if _base in ("status", "version", "health", "agents", "dags", "approvals"):
            _resp = _read_cmd_response(_base)
            log(f"💬 只读命令响应[{_base}]: {_resp[:100]}")
            _api_post("/api/v3/activities/log", {
                "action": "feishu_cmd_response", "agent_id": AGENT_ID,
                "summary": f"cmd={_base} resp={_resp[:200]}", "source": "ws_feishu_handler",
            })
            return

        # ─── 写命令: 长庚总open_id校验(Default-deny) ───
        _cg_oid = os.getenv("CG_OPEN_ID", "")
        if _cg_oid and open_id and open_id != _cg_oid:
            log(f"⛔ [命令桥] 无权限: cmd={_base} open_id={open_id[:10]}...")
            _api_post("/api/v3/activities/log", {
                "action": "feishu_cmd_denied", "agent_id": AGENT_ID,
                "summary": f"cmd={_base} open_id={open_id[:10]} 无权限", "source": "ws_feishu_handler"})
            return

        # T1运维命令(幂等低危): requeue/restart/trigger/cleanup
        if _base in ("requeue", "restart", "trigger", "cleanup"):
            log(f"💬 [命令桥] T1运维命令: {_base} operator={operator}")
            _exec_write_cmd(_base)
            return

        # ─── T2管理命令: 二次确认门(高危) ───
        if _base in ("approve", "reject", "upgrade", "deploy"):
            if not confirm:
                log(f"🔒 [命令桥] {_base}需要确认, 请输入confirm=true执行")
                _write_shared_memory("feishu_cmd", f"{_base}_confirm", {
                    "success": False, "method": "ws_feishu_handler_confirm_required",
                    "output": f"命令[{_base} {_arg}]需二次确认: 请回复confirm=true"})
                return
            _exec_t2_admin_cmd(_base, _arg)
            return

        # 未知命令
        log(f"💬 未知命令[{cmd_lower}], 支持: status/version/health/agents/dags/approvals/"
            f"requeue/restart/trigger/cleanup/approve/reject/upgrade/deploy")
        _write_shared_memory("feishu_cmd", cmd_lower, {
            "success": False, "method": "ws_feishu_handler_unknown",
            "output": f"未知命令[{cmd_lower}]"})
    threading.Thread(target=_exec, daemon=True).start()


def _read_cmd_response(base: str) -> str:
    """命令桥T0只读命令响应"""
    try:
        if base == "version":
            _v = _get_version()
            return f"[天枢] v{_v} | handlers={len(HANDLERS)} | hermes={'OK' if _hermes_available() else 'MISSING'}"
        if base == "status":
            _v = _get_version()
            return f"[天枢] v{_v} online | WS={bool(_CURRENT_WS)} | hermes={'OK' if _hermes_available() else 'MISSING'}"
        if base == "health":
            _v = _get_version()
            return f"[天枢] health=OK v{_v} | handlers={len(HANDLERS)}"
        if base == "agents":
            resp = _api_get("/api/v3/hub/agents")
            items = resp.get("data", {}).get("agents", []) if isinstance(resp, dict) else []
            if isinstance(items, dict):
                items = items.get("agents", [])
            online = sum(1 for a in items if a.get("ws_online"))
            return f"[天枢] Agent在线 {online}/{len(items)}"
        if base == "dags":
            resp = _api_get("/api/v3/workflows/dags")
            d = resp.get("data") if isinstance(resp, dict) else None
            items = d.get("dags", d.get("items", [])) if isinstance(d, dict) else []
            return f"[天枢] DAG共 {len(items)} 个"
        if base == "approvals":
            resp = _api_get("/api/v3/ai-bindings?status=pending&size=10")
            items = resp.get("data") if isinstance(resp, dict) else []
            if isinstance(items, dict):
                items = items.get("items", items.get("bindings", []))
            return f"[天枢] 待审批 {len(items)} 个"
    except Exception as e:
        log(f"⚠️ 只读命令异常[{base}]: {e}")
    return f"[天枢] {base}查询失败"


def _exec_t2_admin_cmd(base: str, arg: str):
    """命令桥T2管理命令执行(已过确认门): approve/reject/upgrade/deploy"""
    try:
        if base == "approve" and arg:
            # 按名字查找pending审批并批准
            resp = _api_get("/api/v3/ai-bindings?status=pending&size=50")
            items = resp.get("data") if isinstance(resp, dict) else []
            if isinstance(items, dict):
                items = items.get("items", items.get("bindings", []))
            target = None
            for it in items:
                if isinstance(it, dict) and arg in (it.get("name", "") or it.get("agent_id", "")):
                    target = it
                    break
            if target:
                _api_post(f"/api/v3/approvals/{target.get('approval_id')}/decide",
                          {"status": "approved", "comment": "命令桥T2批准(长庚总确认)"})
                log(f"✅ [命令桥] approve {arg} 成功")
            else:
                log(f"⚠️ [命令桥] 未找到待审批: {arg}")
        elif base == "reject" and arg:
            resp = _api_get("/api/v3/ai-bindings?status=pending&size=50")
            items = resp.get("data") if isinstance(resp, dict) else []
            if isinstance(items, dict):
                items = items.get("items", items.get("bindings", []))
            target = next((it for it in items if isinstance(it, dict)
                           and arg in (it.get("name", "") or it.get("agent_id", ""))), None)
            if target:
                _api_post(f"/api/v3/approvals/{target.get('approval_id')}/decide",
                          {"status": "rejected", "comment": "命令桥T2驳回(长庚总确认)"})
                log(f"✅ [命令桥] reject {arg} 成功")
            else:
                log(f"⚠️ [命令桥] 未找到待审批: {arg}")
        elif base == "upgrade":
            # 触发全量版本升级广播(发布升级令帖)
            _api_post("/api/v3/forum/posts", {
                "title": f"【升级令】全体AI请升级到最新版本 V{_get_version()}",
                "content": (f"@全体AI 请执行 tianshu-self-upgrade.sh 升级到 V{_get_version()}\n"
                            f"接入文档: https://x.kddauto.com/connect.md"),
                "category": "action", "author": AGENT_ID,
            })
            log("✅ [命令桥] upgrade 升级令已发布")
        elif base == "deploy" and arg:
            _api_post("/api/v3/forum/posts", {
                "title": f"【部署令】发布版本 {arg}",
                "content": f"@全体AI 版本 {arg} 已发布, 请升级。接入文档: https://x.kddauto.com/connect.md",
                "category": "action", "author": AGENT_ID,
            })
            log(f"✅ [命令桥] deploy {arg} 部署令已发布")
        _api_post("/api/v3/activities/log", {
            "action": "feishu_cmd_executed", "agent_id": AGENT_ID,
            "summary": f"cmd={base} arg={arg} executed", "source": "ws_feishu_handler_t2"})
    except Exception as e:
        log(f"⚠️ [命令桥] T2命令执行异常 {base}: {e}")
        _write_shared_memory("feishu_cmd", base, {
            "success": False, "method": "ws_feishu_handler_t2",
            "output": f"执行异常: {str(e)[:100]}"})


def _exec_write_cmd(cmd_lower: str):
    """命令桥T1写命令执行(豆包#24036): requeue/restart/trigger/cleanup
    只读系统状态+幂等操作, 不执行高危命令(高危需0153确认门)。
    """
    try:
        if cmd_lower == "requeue":
            # 批量重放failed任务(对标Temporal DLQ)
            resp = _api_post("/api/v3/task-runs/batch-requeue", {"status": "failed"})
            log(f"💬 [命令桥] requeue完成: {str(resp)[:100]}")
        elif cmd_lower == "cleanup":
            # 清理dispatch积压
            resp = _api_post("/api/v3/dispatch/cleanup", {"older_than_hours": 2})
            log(f"💬 [命令桥] cleanup完成: {str(resp)[:100]}")
        elif cmd_lower == "restart":
            # 重启当前移动端WS daemon(幂等, systemd自动拉起)
            log("💬 [命令桥] restart: 记录意图, 由systemd侧处理(不直接kill避免踢链)")
            _write_shared_memory("feishu_cmd", "restart", {
                "success": True, "method": "ws_feishu_handler",
                "output": "restart意图已记录, 将在安全窗口重启"})
        elif cmd_lower == "trigger":
            # 触发一次巡检
            _api_post("/api/v3/activities/log", {
                "action": "feishu_cmd_trigger_patrol", "agent_id": AGENT_ID,
                "summary": "命令桥触发巡检", "source": "ws_feishu_handler"})
            _patrol_trigger()
            log("💬 [命令桥] trigger: 已触发巡检")
        _api_post("/api/v3/activities/log", {
            "action": "feishu_cmd_executed", "agent_id": AGENT_ID,
            "summary": f"cmd={cmd_lower} executed", "source": "ws_feishu_handler"})
    except Exception as e:
        log(f"⚠️ [命令桥] 写命令执行异常 {cmd_lower}: {e}")
        _write_shared_memory("feishu_cmd", cmd_lower, {
            "success": False, "method": "ws_feishu_handler",
            "output": f"执行异常: {str(e)[:100]}"})


def _patrol_trigger():
    """命令桥trigger: 立即触发一次巡检(复用handle_patrol逻辑)"""
    try:
        handle_patrol({"agent_id": AGENT_ID, "enforce": False})
        log("💬 [命令桥] 巡检已触发(handle_patrol)")
    except Exception as e:
        log(f"⚠️ [命令桥] 巡检触发异常: {e}")

def _feishu_notify(text: str):
    """飞书Webhook推送(长庚总审批通知/告警等)
    环境变量FEISHU_WEBHOOK配置; 未配置则仅日志(论坛帖/共享记忆已兜底)。
    """
    webhook = os.getenv("FEISHU_WEBHOOK", "")
    if not webhook:
        log("⚠️ 未配置FEISHU_WEBHOOK，跳过飞书推送(论坛帖已兜底)")
        return
    try:
        payload = json.dumps({
            "msg_type": "text",
            "content": {"text": text[:4000]},
        }).encode()
        req = urllib.request.Request(webhook, data=payload,
            headers={"Content-Type": "application/json"}, method="POST")
        urllib.request.urlopen(req, timeout=5)
        log(f"✅ 飞书推送成功: {text[:50]}")
    except Exception as e:
        log(f"⚠️ 飞书推送失败: {e}")

# ═══════════════════════════════════════════
# 辅助函数
# ═══════════════════════════════════════════

def send_status(status: str, extra: dict = None):
    try:
        payload = {"agent_id": AGENT_ID, "status": status, "timestamp": time.time()}
        if extra: payload["extra"] = extra
        _api_post("/api/v3/agents/status", payload)
    except Exception as e:
        log(f"[EXCEPTION] Unexpected error in auto-upgrade/daemon cycle: {traceback.format_exc()}")
        pass
    except Exception:
        log(f"[EXCEPTION] Unexpected error: {traceback.format_exc()}")
        pass

def write_trigger(data: dict):
    try:
        with open(TRIGGER_FILE, "w") as f:
            json.dump({"type": data.get("type",""), "received_at": time.time(),
                       "agent_id": AGENT_ID, "data": data}, f)
    except Exception as e:
        log(f"[EXCEPTION] Unexpected error in auto-upgrade/daemon cycle: {traceback.format_exc()}")
        pass
    except Exception:
        log(f"[EXCEPTION] Unexpected error: {traceback.format_exc()}")
        pass

def _trigger_cron():
    global _last_trigger_time
    with _exec_lock:
        now = time.time()
        if now - _last_trigger_time < TRIGGER_COOLDOWN: return
        _last_trigger_time = now
    log("⚡ 触发cron")

def _notify_bridge(data: dict):
    try:
        payload = json.dumps({
            "type": "consensus_alert", "post_id": data.get("post_id",""),
            "title": data.get("title","")[:80], "agent_id": AGENT_ID,
            "timestamp": time.time()
        }).encode()
        req = urllib.request.Request("http://localhost:18998/api/bridge/memory",
            data=payload, headers={"Content-Type": "application/json"}, method="POST")
        urllib.request.urlopen(req, timeout=3)
    except Exception as e:
        log(f"[EXCEPTION] Unexpected error in auto-upgrade/daemon cycle: {traceback.format_exc()}")
        pass
    except Exception:
        log(f"[EXCEPTION] Unexpected error: {traceback.format_exc()}")
        pass

# ★V12.6.0017: 监控回传补全 — 玄机三问落地
def _report_execution(source_type: str, source_id: str, result: dict):
    """每次hermes执行后上报到天枢agent_activity_log
    修复就绪度检测缺口: Agent本地hermes执行结果需回传天枢
    """
    if not TIANSHU_API_KEY or not source_id:
        return
    try:
        send_status("executed", {
            "source_type": source_type,
            "source_id": str(source_id),
            "success": result.get("success", False),
            "method": result.get("method", "unknown"),
            "output_preview": str(result.get("output", ""))[:200],
            "attempts": result.get("attempts", 1)
        })
        log(f"📊 执行上报: {source_type}#{source_id} success={result.get('success')}")
    except Exception as e:
        log(f"⚠️ 执行上报异常: {e}")

def _write_shared_memory(source_type: str, source_id: str, result: dict):
    """任务完成后写入天枢共享记忆
    修复共享记忆缺口: Agent执行结果不可见→天枢无法了解Agent实际执行状态
    V12.6.0014: 统一走bridge门禁(content≥20字符)
    ★ 开发负责人: 天枢移动端 · V12.8.0229 · 2026-08-19 · 长庚总指令"天枢接入升级,充分利用记忆总线"
      记忆总线迁移: 旧端点/memory/shared/write(agent_id/summary/content) → 契约端点/shared-memory/write(key/value/namespace/tags)
      同表(MemoryBus)无缝迁移; key唯一化防409; namespace=shared
    """
    if not TIANSHU_API_KEY or not source_id:
        return
    try:
        success = result.get("success", False)
        output = str(result.get("output", ""))[:300]
        summary = f"[{AGENT_ID}] {source_type}#{source_id} {'✅成功' if success else '❌失败'}"
        content = f"来源: {source_type}#{source_id}\nAgent: {AGENT_ID}\n方法: {result.get('method','?')}\n结果: {'成功' if success else '失败'}\n输出: {output}\n时间: {time.strftime('%Y-%m-%d %H:%M:%S')}"
        _api_post("/api/v3/shared-memory/write", {
            "key": f"daemon:{source_type}:{source_id}",
            "value": content,
            "namespace": "shared",
            "memory_type": "short_term",
            "tags": [source_type, "daemon"],
            "source": f"agent_loop_ws_v4:{AGENT_ID}"
        })
        log(f"📝 共享记忆写入: {summary[:80]}")
    except Exception as e:
        log(f"⚠️ 共享记忆写入异常: {e}")

# ═══════════════════════════════════════════
# 消息路由器
# ═══════════════════════════════════════════

# ★V12.8.0269(豆包#28406 P0-C): 事件DLQ — handler失败落盘(对标AWS SQS DLQ)
#   根因: 原fire-and-forget子线程+主线程立即_msg_id_complete → handler实际失败也被标记成功(假完成)
#   修复: handler安全执行包装(_run_handler_safe): 成功→complete+result ACK;
#         失败→写DLQ+error ACK(Hub可感知), 不complete(WAL不记) → 连接断开后resume重放=自然重试
#   精简: JSONL append-only, 每行一个失败事件; 启动时_dlq_replay重放attempts<3
#   上限(ponytail): DLQ文件无限增长 — 若失败>200条/日需归档轮转, 当前低频场景够用
_DLQ_FILE = f"/tmp/agent_ws_dlq_{AGENT_ID}.jsonl"
_DLQ_MAX_ATTEMPTS = 3


def _dlq_append(msg_type: str, msg: dict, exc: Exception):
    """handler失败事件写入DLQ(JSONL append-only)"""
    try:
        rec = {"ts": datetime.datetime.now().isoformat(), "msg_type": msg_type,
               "mid": (msg or {}).get("message_id", ""), "error": str(exc)[:200],
               "traceback": traceback.format_exc()[-2000:], "attempts": 1,
               "msg": json.dumps(msg, ensure_ascii=False)[:5000]}
        with open(_DLQ_FILE, "a", encoding="utf-8") as _df:
            _df.write(json.dumps(rec, ensure_ascii=False) + "\n")
    except Exception as _de:
        log(f"[DLQ] 写入失败: {_de}")


def _dlq_replay():
    """启动时重放DLQ: attempts<3重新入队(attempts+1), >=3保留审计"""
    if not os.path.exists(_DLQ_FILE):
        return
    try:
        kept = []
        with open(_DLQ_FILE, "r", encoding="utf-8") as _df:
            for _line in _df:
                _line = _line.strip()
                if not _line:
                    continue
                try:
                    _rec = json.loads(_line)
                except Exception:
                    continue
                if _rec.get("attempts", 0) < _DLQ_MAX_ATTEMPTS:
                    _rec["attempts"] = _rec.get("attempts", 0) + 1
                    try:
                        _msg = json.loads(_rec.get("msg", "{}"))
                        _mt = _rec.get("msg_type", "")
                        if _mt in HANDLERS:
                            log(f"[DLQ] 重放 #{_rec['attempts']}: {_mt} mid={str(_rec.get('mid',''))[:8]}")
                            threading.Thread(target=_run_handler_safe,
                                             args=(_mt, _msg, None, _rec.get("mid", ""), ""),
                                             daemon=True).start()
                    except Exception as _re:
                        log(f"[DLQ] 重放解析失败: {_re}")
                kept.append(json.dumps(_rec, ensure_ascii=False))
        with open(_DLQ_FILE, "w", encoding="utf-8") as _df:
            _df.write("\n".join(kept) + ("\n" if kept else ""))
        log(f"🔁 DLQ重放完成: 保留{len(kept)}条(重放/审计)")
    except Exception as _e:
        log(f"⚠️ DLQ重放失败: {_e}")


def _run_handler_safe(msg_type: str, msg: dict, ws, mid: str, tid: str):
    """★V12.8.0269(豆包#28406 P0-C): handler安全执行 — 成功才complete, 失败进DLQ
    根因: 原fire-and-forget子线程+主线程立即complete → handler失败也被标记成功(假完成, at-most-once)
    修复: 成功→complete+result ACK; 失败→DLQ落盘+error ACK(Hub台账可见失败), 不complete
    """
    try:
        HANDLERS[msg_type](msg)
        if mid:
            _msg_id_complete(mid)
            _send_ack(ws, mid, "result", tid, result="handled")
    except Exception as _e:
        _dlq_append(msg_type, msg, _e)
        log(f"[DLQ] handler执行失败 msg_type={msg_type} mid={mid[:8] if mid else '?'}: {traceback.format_exc()}")
        if mid:
            _send_ack(ws, mid, "result", tid, result=f"error:{type(_e).__name__}")


def route_message(msg: dict, ws=None):
    global _last_business_msg  # ★V12.6.0035: WS僵死自愈 — 业务消息跟踪
    msg_type = msg.get("type", "")
    # ★V12.6.0035: 排除ping/pong/heartbeat — 仅业务消息刷新时间戳
    if msg_type not in ("ping", "pong", "heartbeat"):
        _last_business_msg = time.time()

    # ★V12.8.0115(豆包顾问P0#3): 处理服务端ACK(server_received)
    #   根因: Agent→天枢单向消息无确认 → Agent不确定送达 → 重发5-9次(豆包实测ACK率28.5%)
    #   修复: 天枢回 {"type":"ack","ack_stage":"server_received"} → Agent确认送达, 记录并停止重发
    if msg_type == "ack" and msg.get("ack_stage") == "server_received":
        _ack_mid = msg.get("message_id", "")
        if _ack_mid:
            with _DELIVERED_MSG_LOCK:
                if _ack_mid not in _DELIVERED_MSG_IDS:
                    _DELIVERED_MSG_IDS.append(_ack_mid)
                    if len(_DELIVERED_MSG_IDS) > _DELIVERED_MSG_IDS_MAX:
                        del _DELIVERED_MSG_IDS[:len(_DELIVERED_MSG_IDS) - _DELIVERED_MSG_IDS_MAX]
        log(f"✅ 服务端确认送达: mid={_ack_mid[:12] if _ack_mid else '?'}")
        return

    # [kimiK3] V12.8.0016: 三段式ACK+message_id幂等(E1101重复投递根治)
    # 协议: transport(到达)→handler(开始执行)→result(执行完成)
    _mid = msg.get("message_id", "")
    _tid = msg.get("task_id", "") or msg.get("id", "")
    # ★V12.8.0183批次2(kimik3): 跟踪last_seq(断线resume游标)——单调递增, 降频持久化
    _seq = msg.get("seq", 0)
    if _seq:
        with _WS_SESSION_LOCK:
            if _seq > _WS_SESSION.get("last_seq", 0):
                _WS_SESSION["last_seq"] = _seq
        if _seq % 20 == 0:  # 每20条写盘一次, 防每次IO
            _save_ws_session()
    if _mid and msg_type not in ("ping", "pong", "heartbeat", "ack"):
        _send_ack(ws, _mid, "transport", _tid)
        # ★V12.8.0186(数据架构师109727): 幂等标记时机 — begin(仅内存)而非立即标记processed
        #   根因: 原_msg_id_seen接收时即标记+WAL持久化 → 执行中崩溃=消息丢失(at-most-once)
        #   修复: begin只标记进行中(防并发重复), 执行完成后_msg_id_complete才写WAL(at-least-once)
        if _msg_id_begin(_mid):
            # 重复消息(已完成/进行中): 回result ACK让Hub停止重投，但不重复执行
            log(f"↩️ 重复消息已幂等去重: {msg_type} mid={_mid[:8]}")
            _send_ack(ws, _mid, "result", _tid, result="duplicate_ignored")
            return

    if msg_type in HANDLERS:
        if _mid:
            _send_ack(ws, _mid, "handler", _tid)
        # ★V12.8.0267(豆包#28406 P0-A): handler在子线程执行, 不阻塞ws.recv循环
        #   根因: handler同步调_hermes_execute(最长120s+) → 阻塞recv → pong无法及时回复
        #         → 服务端45s判离线 → 重连风暴(豆包评分4.5/10核心根因)
        #   修复: 子线程执行handler(与MCP巡检同模式), recv循环保持响应pong
        # ★V12.8.0269(豆包#28406 P0-C): 改_run_handler_safe — 成功才complete, 失败进DLQ
        threading.Thread(target=_run_handler_safe, args=(msg_type, msg, ws, _mid, _tid), daemon=True).start()
    elif "post_id" in msg and "title" in msg:
        if _mid:
            _send_ack(ws, _mid, "handler", _tid)
        # ★V12.8.0269(P0-C): fallback分支同样走安全执行(原同步调用阻塞+无DLQ)
        threading.Thread(target=_run_handler_safe, args=("consensus_alert", msg, ws, _mid, _tid), daemon=True).start()
    elif msg_type == "ping":
        # ★V12.3.1修复: Hub的application-level ping必须回复pong
        if ws:
            try:
                asyncio.create_task(ws.send(json.dumps({"type": "pong"})))
            except Exception:
                # ★V12.8.0187(数据架构师109774): 重复except死代码删除 —
                #   根因: 原try后有两个except Exception(第一个已捕获全部→第二个永不执行)
                #         +日志描述从daemon循环复制粘贴错误("auto-upgrade/daemon cycle"与ping无关)
                #   修复: 保留单个except+修正日志描述(pong回发失败, 连接将断开重连)
                log(f"[EXCEPTION] pong回发失败: {traceback.format_exc()}")
        return
    elif msg_type in ("heartbeat", "pong"):
        return
    else:
        log(f"  ← {msg_type}: {str(msg)[:100]}")

    # ★V12.8.0269(P0-C): result ACK/complete 移至 _run_handler_safe 内部(成功才complete)
    #   根因: 原主线程立即complete → handler线程实际失败也被标记成功(假完成)
    #   现在: handler成功→complete+result ACK; 失败→DLQ+error ACK, 不complete

# ═══════════════════════════════════════════
# WS主循环
# ═══════════════════════════════════════════

# ★V12.8.0186(批次3·6态连接状态机): 全局连接状态(IDLE→CONNECTING→CONNECTED→
#   RESUMING→RECONNECTING→STOPPED), 供ws-health/可观测性查询当前连接阶段
_CONN_STATE = "IDLE"


async def listen_ws():
    # ★V12.8.0023: 启动自检门禁 — 配置错误在启动时暴露，不再运行时循环403
    if not AGENT_ID or AGENT_ID.strip().lower() == "unknown":
        log("❌ 启动拒绝: AGENT_ID未设置或为unknown — 请设置环境变量 AGENT_ID=你的中文名")
        log("   修复: systemd unit加 Environment=\"AGENT_ID=你的Agent名\" 后 systemctl restart")
        sys.exit(2)
    if not TIANSHU_API_KEY.startswith("tsk_") or len(TIANSHU_API_KEY) < 20:
        log(f"❌ 启动拒绝: api_key格式无效(长度{len(TIANSHU_API_KEY)}, 前缀{TIANSHU_API_KEY[:4]!r}) — 必须tsk_开头完整Key")
        log("   修复: curl -X POST https://x.kddauto.com/api/v3/auth/api-keys/auto-register -H 'Content-Type: application/json' -d '{\"agent_id\": \"你的Agent名\"}'")
        log("   警告: 禁止从天枢日志/帖子中复制已脱敏(***)的Key！")
        sys.exit(2)
    # ★V12.8.0119 命名规范铁律: AGENT_ID必须含中文字符+禁止后缀(与agent_name.py同规则)
    if not re.search(r"[\u4e00-\u9fff]", AGENT_ID):
        log(f"❌ 启动拒绝: AGENT_ID '{AGENT_ID}' 非法 — 必须含中文字符(命名规范铁律)")
        log("   修复: 设置 AGENT_ID=你的中文名 (例: AGENT_ID=流音)")
        sys.exit(2)
    if re.search(r"-[A-Za-z0-9\u4e00-\u9fff]+$", AGENT_ID.strip()):
        log(f"❌ 启动拒绝: AGENT_ID '{AGENT_ID}' 非法 — 中文名禁止后缀(-xxx), 厂商/型号走owner参数")
        sys.exit(2)
    # ★V12.8.0119 公网绝对地址铁律: 禁止localhost/内网IP/非https
    if not TIANSHU_API.startswith("https://"):
        log(f"❌ 启动拒绝: TIANSHU_API '{TIANSHU_API}' 非法 — 必须使用https公网绝对地址(https://x.kddauto.com)")
        sys.exit(2)
    if re.search(r"localhost|127\.0\.0\.1|10\.\d+\.\d+\.\d+|192\.168\.", TIANSHU_API):
        log(f"❌ 启动拒绝: TIANSHU_API '{TIANSHU_API}' 包含内网地址 — 公网绝对地址铁律(外部AI一律走公网域名)")
        sys.exit(2)

    # ★V12.8.0115(豆包顾问P0#1): identity端点需携带fingerprint(服务端challenge-response认证)
    ws_url = f"{WS_URL}?api_key={TIANSHU_API_KEY}&agent_id={urllib.parse.quote(AGENT_ID)}"
    if "/ws/agent/identity" in WS_URL:
        _fp = ""
        try:
            _id_cfg = os.path.expanduser(f"~/.tianshu/agent_identity_{AGENT_ID}.json")
            if os.path.exists(_id_cfg):
                import json as _json
                _fp = _json.load(open(_id_cfg)).get("fingerprint", "")
            if not _fp:
                for _cand in ("~/.tianshu/identity.json", "~/.tianshu/agent_identity.json"):
                    _p = os.path.expanduser(_cand)
                    if os.path.exists(_p):
                        import json as _json
                        _fp = _json.load(open(_p)).get("fingerprint", "")
                        break
        except Exception:
            _fp = ""
        if _fp:
            ws_url += f"&fingerprint={urllib.parse.quote(_fp)}"
        else:
            log("⚠️ identity端点未找到本地fingerprint, 回退旧端点(仍可用但无challenge-response)")
            ws_url = ws_url.replace("/ws/agent/identity", "/ws/agent")

    # ★V12.6.0020: 连接即升级 — 启动时强制检查connect.md+自动下载最新版
    _self_upgrade_check()
    
    log(f"V{_get_version()}-SAFE启动 | hermes={'OK' if _hermes_available() else 'MISSING'} | profile={_resolve_profile(AGENT_ID)} | sem={_MAX_CONCURRENT_HERMES} | handlers={list(HANDLERS.keys())}")
    
    # V12.6.0: 启动时同步文档
    _sync_docs()
    
    # V12.6.0: 定时后台任务
    async def periodic_tasks():
        """每60秒执行HTTP心跳 + 文档同步检查 + 健康上报 + MCP任务巡检 + 每30分钟自升级检查"""
        _last_upgrade_check = 0  # V12.6.0033: 定期自升级
        # ★ 开发负责人: 天枢工作区 · V12.8.0246 · 2026-08-21 · 对标Hermes delivery_ledger.sweep_recoverable
        #   根因: inflight_query只在重连时问一次(V12.8.0189), 若WS长连稳定但任务执行中崩溃(进程内异常)
        #         悬挂任务无人回收 — Hermes用sweep_recoverable周期性扫描+恢复, 天枢只靠重连触发=盲区
        #   修复: periodic_tasks加周期性inflight_query(每15分钟), 主动询问服务端pending_acks未完成任务
        _last_inflight_sweep = 0
        while True:
            await asyncio.sleep(60)
            try:
                # ★V12.6.0073: HTTP心跳到Daemon API — 修复daemon_agents表与WS在线脱节(议题#47071)
                threading.Thread(target=_daemon_heartbeat, daemon=True).start()
                # 每小时文档同步
                if time.time() - _last_doc_sync > 3600:
                    threading.Thread(target=_sync_docs, daemon=True).start()
                # 每5分钟健康上报
                threading.Thread(target=_health_report, daemon=True).start()
                # V12.6.0002: 每3分钟MCP任务巡检（拉取看板任务+自主执行）
                threading.Thread(target=_mcp_periodic_scan, daemon=True).start()
                # V12.6.0033: 每30分钟自升级检查（启动时已检查过一次，定期复检）
                if time.time() - _last_upgrade_check > 1800:
                    threading.Thread(target=_self_upgrade_check, daemon=True).start()
                    _last_upgrade_check = time.time()
                # ★对标Hermes sweep_recoverable: 每15分钟周期性询问悬挂任务(不只重连时)
                if time.time() - _last_inflight_sweep > 900 and _CURRENT_WS is not None:
                    try:
                        await _CURRENT_WS.send(json.dumps({
                            "type": "inflight_query", "agent_id": AGENT_ID,
                            "reason": "periodic_sweep"}))
                        log("🔄 周期性悬挂任务扫描(15min sweep, 对标Hermes delivery_ledger)")
                        _last_inflight_sweep = time.time()
                    except Exception as _ise:
                        log(f"⚠️ 周期性inflight_query失败(不影响连接): {_ise}")
            except Exception as _pt_we:
                # ★V12.8.0186: periodic_tasks catch-all — 保留防崩溃但需可观测(否则线程spawn失败被吞=Agent僵死)
                log(f"⚠️ periodic_tasks循环异常: {_pt_we}")
    
    # ★V12.8.0187(履约109785 Bug#4): backoff死变量移除(Full Jitter取代) —
    #   原`backoff = 1`初始化+`backoff=min(backoff*2,30)`更新均无读取点=死代码
    reconnect_count = 0  # V12.6.0031: 重连计数，用于日志+Full Jitter退避
    periodic_task = None
    # ★V12.8.0184(P1-7·豆包109687 P1#5): ping_interval吃服务端server_config下发
    #   根因: 原硬编码30, 服务端auth_ok下发server_config.ping_interval=30但客户端不读
    #   修复: SERVER_CONFIG优先, 服务端下发值30s(websockets库默认值/75%规则45s阈值内)
    # ★V12.8.0187(豆包109791 P1-3): 默认值对齐V3.0裁定 20s/10s —
    #   根因: 原默认30s/33.75s(45*0.75), 总死连接检测63.75s是裁定30s两倍多;
    #         服务端server_config可覆盖, 但默认值应直接设对(豆包: "默认值应直接设对")
    _ws_ping_interval = SERVER_CONFIG.get("ping_interval", 20)      # V3.0裁定20s(发Ping间隔)
    _ws_ping_timeout = SERVER_CONFIG.get("ping_timeout", 10)        # V3.0裁定10s(等Pong超时)
    # ★V12.8.0186(批次3·V3.0治理层): 6态连接状态机 — 可观测性(ws-health看到Agent当前状态)
    #   状态: IDLE(空闲) → CONNECTING(连接中) → CONNECTED(已连接)
    #         → RESUMING(恢复会话) → RECONNECTING(重连中) → STOPPED(停止)
    #   对标: Discord Gateway connection states + K8s Pod phases(可观测性三支柱)
    global _CONN_STATE
    _CONN_STATE = "CONNECTING"
    while True:
        try:
            import websockets
            _CONN_STATE = "CONNECTING"
            async with websockets.connect(ws_url, ping_interval=_ws_ping_interval, ping_timeout=_ws_ping_timeout, close_timeout=5, open_timeout=30) as ws:
                global _CURRENT_WS
                global _CURRENT_LOOP  # ★V12.8.0267(P0-A): 记录主loop供子线程handler
                _CURRENT_WS = ws  # [kimiK3] V12.8.0016: 供handler发送响应
                _CURRENT_LOOP = asyncio.get_running_loop()
                _CONN_STATE = "CONNECTED"
                # ★V12.8.0186(豆包109745 P2#7): TCP keepalive — CPU饥饿保护(SO_KEEPALIVE)
                #   根因: 网络中间设备(防火墙/NAT)静默丢弃空闲连接, 应用层Ping间隔期间半开连接
                #         → 假在线(连接看似存活实则僵死)
                #   依据: RFC 1122 TCP keepalive + Linux tcp_keepalive_time默认7200s
                #   修复: 连接建立后设SO_KEEPALIVE(OS层, 2h→75s探测, 与应用层Ping 20s互补)
                try:
                    _tr = getattr(ws, "transport", None)
                    if _tr is not None:
                        _sock = _tr.get_extra_info("socket")
                        if _sock is not None:
                            _sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
                            try:
                                _sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 75)
                                _sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, 30)
                                _sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPCNT, 3)
                            except OSError:
                                pass  # 部分平台无TCP_KEEPIDLE等常量(Windows), 保留默认keepalive
                except Exception as _kae:
                    log(f"⚠️ TCP keepalive设置失败(不影响连接): {_kae}")
                # V12.6.0031: 增强重连日志，显示重连次数
                if reconnect_count > 0:
                    log(f"✅ WS已连接 (第{reconnect_count}次重连成功)")
                else:
                    log("✅ WS已连接")
                reconnect_count = 0  # V12.6.0031: 连接成功重置计数(Full Jitter退避归零)
                _last_business_msg = time.time()  # ★V12.6.0085: 重连后重置僵死检测计时器, 防止立即重新触发断开
                # V12.6.0: 启动后台定时任务
                if periodic_task is None:
                    periodic_task = asyncio.create_task(periodic_tasks())
                # ★ V12.8.0258(天枢移动端·帖#28415): 飞书WS长连接启动(审批回调专用，与天枢WS并行)
                #   依赖: FEISHU_APP_ID/FEISHU_APP_SECRET环境变量
                #   ponytail: 模块已就绪，4行启动代码
                _feishu_ws_started = False
                try:
                    if _FEISHU_WS_AVAILABLE and os.getenv("FEISHU_APP_ID") and os.getenv("FEISHU_APP_SECRET"):
                        _approval_h = FeishuApprovalHandler()
                        _feishu_client = FeishuWSClient(
                            os.getenv("FEISHU_APP_ID"),
                            os.getenv("FEISHU_APP_SECRET"),
                            on_card_action=_approval_h.handle_approval,
                        )
                        _feishu_client.start()
                        _feishu_ws_started = True
                        log("✅ 飞书WS长连接已启动")
                except Exception as _fwe:
                    log(f"⚠️ 飞书WS启动失败(不影响主连接): {_fwe}")
                # ★V12.8.0115-fix(豆包顾问第五轮P0): identity端点先做Ed25519 challenge-response认证
                #   根因: SDK只改URL没写认证逻辑 → 服务端发challenge不响应 → 认证超时断开 → 新端点废
                if "/ws/agent/identity" in ws_url:
                    _priv_key, _fp = _load_identity_key()
                    if _priv_key and _fp:
                        _auth_ok = await _auth_identity(ws, _priv_key, _fp)
                        if not _auth_ok:
                            log("⚠️ identity认证失败, 连接将被服务端关闭, 重连")
                            raise ConnectionError("identity auth failed")
                    else:
                        log("⚠️ 本地无私钥/指纹, identity端点将认证失败")
                        # 无私钥时不升级, 但继续尝试(服务端可能对tsk_self_internal免认证)
                try:
                    # [kimiK3] V12.8.0016: capabilities动态生成(阶段3.4)
                    #   根因: 原39项硬编码与实际能力无关——handler注册表才是真实能力
                    _real_caps = sorted(set(
                        ["ws-daemon", "hermes-bridge", "poll-engine"] + list(HANDLERS.keys())
                    ))
                    await ws.send(json.dumps({
                        "type": "register", "agent_id": AGENT_ID,
                        "capabilities": _real_caps,
                        "version": _get_version(), "runtime_type": "python-ws-daemon-v4",
                        "sdk_md5": _sdk_md5(),  # [kimiK3] 阶段3.1: 内容哈希握手(治同名不同内容碎片化)
                        "daemon_version": _get_daemon_version(),  # [kimiK3] V12.8.0017: daemon版本空间(4.x)上报——Hub双因子判定主因子(帖#22450 force_update假阳性根治)
                        "hermes_version": _get_hermes_version(),  # ★V12.6.0027: daemon上报真实Hermes CLI版本
                        "hermes_bridge_ok": _hermes_available(),   # ★V12.6.0027: daemon上报桥接可用状态
                        # ★V12.8.0190(T3·玄机109868实证): ws_running=真实运行中 —
                        #   根因(R3): 改码不重启→运行进程仍是旧版(无新handler)→WS升级通道失效
                        #   修复: 上报运行状态, 服务端可区分"进程在跑"vs"代码最新"(guardian兜底重启)
                        "ws_running": True,
                    }))
                    # ★V12.8.0183批次2(kimik3): 断线重连且有保存会话→发resume请求恢复丢失消息
                    #   (Discord Gateway Resume语义; 服务端epoch校验+token验签+按序重放seq>last_seq)
                    with _WS_SESSION_LOCK:
                        _rs = dict(_WS_SESSION)
                    if _rs.get("session_id") and _rs.get("resume_token"):
                        try:
                            # ★V12.8.0186(批次3·6态状态机): 发resume时标记RESUMING
                            _CONN_STATE = "RESUMING"
                            await ws.send(json.dumps({
                                "type": "resume", "session_id": _rs["session_id"],
                                "last_seq": _rs.get("last_seq", 0),
                                "epoch": _rs.get("epoch", ""),
                                "resume_token": _rs.get("resume_token", ""),
                            }))
                            log(f"🔄 发送resume请求: sid={_rs['session_id'][:8]}... last_seq={_rs.get('last_seq',0)}")
                        except Exception as _re:
                            log(f"⚠️ resume请求发送失败: {_re}")
                except Exception as e:
                    # ★V12.8.0117-fix(豆包#67823 P1): 删除多余第二个except块(永不执行)
                    log(f"[EXCEPTION] register发送失败: {traceback.format_exc()}")
                
                while True:
                    try:
                        msg = await asyncio.wait_for(ws.recv(), timeout=90)
                        data = json.loads(msg)
                        route_message(data, ws)
                        # ★V12.6.0035: WS僵死自愈 — 每次业务消息后检查是否>10min零业务消息
                        # ★V12.8.0268(天枢·豆包#28406 P1-B): zombie阈值参数化 — 服务端zombie_timeout可覆盖
                        if time.time() - _last_business_msg > SERVER_CONFIG.get("zombie_timeout", BUSINESS_MSG_TIMEOUT):
                            log(f"⚠️ WS僵死检测: 已{time.time()-_last_business_msg:.0f}s零业务消息 → 主动断开重连")
                            break  # 跳出内层循环 → 触发WS重连
                    except asyncio.TimeoutError:
                        try: await ws.send(json.dumps({"type": "ping"}))
                        except Exception:
                            log(f"[EXCEPTION] Loop break on error: {traceback.format_exc()}")
                            break
                        except Exception:
                            log(f"[EXCEPTION] Loop error: {traceback.format_exc()}")
                            break
                        # ★V12.6.0035: WS僵死自愈 — 90s无消息超时也检查僵死
                        # ★V12.8.0268(天枢·豆包#28406 P1-B): zombie阈值参数化 — 服务端zombie_timeout可覆盖
                        if time.time() - _last_business_msg > SERVER_CONFIG.get("zombie_timeout", BUSINESS_MSG_TIMEOUT):
                            log(f"⚠️ WS僵死检测(超时触发): 已{time.time()-_last_business_msg:.0f}s零业务消息 → 主动断开重连")
                            break
        except Exception as e:
            # V12.6.0031: 指数退避+jitter防雷同重连，上限60→30加速恢复
            # V12.8.0080-FIX: 对称抖动(对标Multica jitterDuration ±50%)
            #   旧实现 uniform(0, 0.3*backoff) 是单向(只加不减) → 系统性延迟重连
            #   Multica用 ±50%对称抖动打破惊群且不延迟恢复
            # ★V12.8.0183批次2(kimik3): Full Jitter退避, 吃服务端ClientConfig(auth_ok下发)
            #   依据AWS Architecture Blog: sleep=random(0,min(cap,base*2^attempt)) 治重连风暴惊群
            #   替代旧"指数退避+对称±50%抖动"——Full Jitter随机性更均匀, 多Agent同时断连不共振
            _rb = SERVER_CONFIG.get("reconnect_base_ms", 1000) / 1000.0
            _rc = SERVER_CONFIG.get("reconnect_cap_ms", 60000) / 1000.0
            jitter = random.uniform(0, min(_rc, _rb * (2 ** max(0, reconnect_count))))
            wait = max(0.5, jitter)
            reconnect_count += 1
            # ★V12.8.0090: WS被拒时提示审批链接(审批全流程显式化·长庚总指令)
            #   根因: AI装完脚本看到"接入完成"就以为接入了, 实际 is_active=false 时WS被拒(4001)却不知原因
            _e_str = str(e)
            if ("4001" in _e_str or "未激活" in _e_str or "不存在" in _e_str or
                    "not activated" in _e_str.lower() or "Connection closed" in _e_str):
                log("⛔ WS连接被拒 — 可能原因: 本AI尚未通过长庚总审批激活")
                log("   审批是唯一上线路径: 请先完成接入申请并等待长庚总在 https://x.kddauto.com/ai-approvals 批准")
                log("   状态自查: curl -s https://x.kddauto.com/api/v3/agents/apply/status/<apply_id>")
                log(f"   详情: {_e_str[:200]}")
            # ★V12.8.0184(P0-2·豆包109687 P0#2): 4001/4002/4005关闭码→停止秒级无限重连
            #   根因: 灵枢实测——服务端明确拒绝(close code)时客户端仍1秒级无限重连=重连风暴
            #   依据: Discord官方close codes——"应把某些close code视为停止重连信号"
            #        4001=拒绝连接(认证/未激活) 4002=协议错误 4005=版本淘汰
            #   修复: 明确拒绝码→300s慢重连(5分钟, 等待长庚总审批/人工介入后自动续连)
            # ★V12.8.0186(批次3·V3.0治理层): 补齐语义码4003-4008分类处理 —
            #   4003=会话失效(重放/epoch不匹配) 4004=无效帧/协议违约 4006=连接冷却(429对等)
            #   4007=Agent已迁移(新服务器接管, 本实例应停止) 4008=服务器重启(短期可自动恢复)
            #   分类: 4001/4002/4003/4005/4007=慢重连或停止; 4004=协议违约5s重试;
            #         4006=按服务端Retry-After冷却; 4008=服务器重启短退避即可(无需300s)
            _close_code_rejected = any(c in _e_str for c in ("4001", "4002", "4003", "4005", "4007", "InvalidStatus"))
            if _close_code_rejected:
                if "4007" in _e_str:
                    # ★V12.8.0186-FIX(豆包109791 P1-1/履约专家109785 Bug3): 4007先设状态再return —
                    #   根因: 原return后L2781-2782的_CONN_STATE="STOPPED"是死代码(执行不到),
                    #         状态机停在RECONNECTING→ws-health看不到"已迁移"真实状态
                    #   修复: return前设STOPPED(状态机终态, 可观测)
                    _CONN_STATE = "STOPPED"  # 已迁移: 状态机终态(不再重连)
                    log(f"⛔ 服务端告知本Agent已迁移至其他服务器 — 停止重连(4007 MIGRATED): {_e_str[:120]}")
                    return  # 已迁移: 本实例不再重连(由新服务器接管)
                log(f"⛔ 服务端明确拒绝(close code) — 进入5分钟慢重连(防重连风暴): {_e_str[:120]}")
                wait = 300
                reconnect_count = 0  # 慢重连重置计数, 防Full Jitter随attempt指数暴涨
            elif "4006" in _e_str:
                # 连接冷却: 服务端要求等待(对标HTTP 429 Retry-After)
                log(f"⏳ 服务端连接冷却(4006) — 30s后重试: {_e_str[:120]}")
                wait = 30
                reconnect_count = 0
            elif "4008" in _e_str:
                # 服务器重启: 短期自动恢复, 5s短退避即可
                log(f"🔄 服务器重启(4008) — 5s后重连: {_e_str[:120]}")
                wait = 5
                reconnect_count = 0
            # ★V12.8.0187(履约109785 Bug#4): 日志去掉backoff(死变量) —
            #   根因: wait已由Full Jitter计算(jitter), backoff不再参与wait但日志仍打印→误导排障
            log(f"⚠️ WS断连 (#{reconnect_count}): {_e_str[:160]}，{wait:.1f}s后重连 (jitter={jitter:.1f}s)")
        else:
            # V12.6.0045: 正常退出(僵死检测break)用固定短退避，修复wait未绑定crash
            wait = 5
        # ★V12.8.0186(批次3·6态状态机): 断连→RECONNECTING(4007已在上面return前设STOPPED)
        #   注: 豆包109791 P1-1指出原L2786-2787死代码(4007在L2761已return, 到不了这里)
        _CONN_STATE = "RECONNECTING"
        await asyncio.sleep(wait)
        # ★V12.8.0187(履约109785 Bug#4): backoff死变量删除 —
        #   根因: Full Jitter(reconnect_count)已取代backoff参与wait, 原`backoff=min(backoff*2,30)`
        #         更新了但无人读取=死代码(还误导日志) — 删除, 保留reconnect_count为唯一退避状态

if __name__ == "__main__":
    cmd = sys.argv[1] if len(sys.argv) > 1 else "listen"
    if cmd == "listen":
        # ★V12.8.0185(小天109718/知微109716/玄机109704 P0#4): flock单实例锁 —
        #   根因: 灵枢实测8个poll进程并发繁殖(3 systemd+容器+孤儿PPID=1), 0184声称修复但代码0处flock
        #   依据: POSIX flock(2)按fd区分非阻塞(LOCK_EX|LOCK_NB), 单实例daemon标准模式(APUE第12章)
        #   机制: /var/lock/tianshu-ws-{AGENT_ID}.lock 非阻塞获取, 失败=已有实例→立即退出(防双实例竞争WS连接)
        _flock_fd = None
        try:
            import fcntl
            _lock_path = f"/var/lock/tianshu-ws-{AGENT_ID}.lock"
            _flock_fd = open(_lock_path, "w")
            fcntl.flock(_flock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except (OSError, IOError, ImportError):
            log("⚠️ flock获取失败(已有实例运行) — 本实例退出, 防双实例竞争WS连接(0185 P0#4)")
            sys.exit(0)
        # ★ V12.8.0118: Key一致性自检(豆包+小天诊断#67901) — 假在线比离线危害更大
        #   对标K8s Startup Probe: 启动检查三处Key(poll.env/ws.env/api_key文件), 不一致即退出
        #   由环境变量TIANSHU_KEY_CHECK=0可跳过(兼容特殊部署)
        if os.getenv("TIANSHU_KEY_CHECK", "1") != "0":
            if not _check_key_consistency():
                log("❌ Key一致性自检失败 — 请统一三处API Key后重启。可通过 TIANSHU_KEY_CHECK=0 跳过。")
                sys.exit(1)
        # V12.6.0042: 守护循环 — 容器环境无systemd时自动重启异常退出的daemon
        # 自升级(os._exit)不走此循环；此循环仅处理非自升级的异常退出
        # ★ 开发负责人: 天枢工作区 · V12.8.0249 · 2026-08-22 · 对标Hermes restart_loop_guard.py(214行)
        #   根因: 原_RESTART_DELAY内存指数退避(3s→60s)在进程崩溃后丢失——systemd Restart=always拉起
        #         的新进程从3s重新开始, 若配置错误导致秒级崩溃循环, 退避永远到不了60s=无限空转
        #   对标: Hermes restart_loop_guard持久化时间戳链(restart_loop.json) → 跨进程识别重启循环
        #   修复: 启动时读重启时间戳链文件, 若近10分钟重启≥5次 → 进入300s长退避(等人工介入)
        _RESTART_DELAY = 3
        _rl_path = f"/tmp/agent_ws_restart_{AGENT_ID}.json"
        try:
            _rl_boots = []
            if os.path.exists(_rl_path):
                with open(_rl_path, "r", encoding="utf-8") as _rlf:
                    _rl_boots = json.loads(_rlf.read()).get("boots", [])
            _now_ts = time.time()
            _rl_boots = [t for t in _rl_boots if _now_ts - float(t) <= 600]  # 保留10分钟内
            _rl_boots.append(_now_ts)
            with open(_rl_path, "w", encoding="utf-8") as _rlf:
                _rlf.write(json.dumps({"boots": _rl_boots}))
            if len(_rl_boots) >= 5:
                _RESTART_DELAY = 300
                log(f"⛔ 重启循环防护触发(对标Hermes restart_loop_guard): 10分钟内{len(_rl_boots)}次重启 → 300s长退避, 疑似配置错误需人工介入")
        except Exception as _rle:
            log(f"⚠️ 重启循环防护初始化失败(不影响运行): {_rle}")
        while True:
            try:
                _dlq_replay()  # ★V12.8.0269(P0-C): 启动时重放DLQ失败事件(attempts<3, 幂等: 文件不存在即跳过)
                asyncio.run(listen_ws())
            except KeyboardInterrupt:
                log("Graceful shutdown")
                break
            except SystemExit:
                # 自升级触发的os._exit直接终止进程，不会到这里
                # 这里的SystemExit是sys.exit触发的，正常退出不重启
                log("Normal exit (sys.exit), 不重启")
                break
            except Exception as e:
                log(f"⚠️ daemon异常退出: {e}，{_RESTART_DELAY}s后自动重启...")
                time.sleep(_RESTART_DELAY)
                _RESTART_DELAY = min(_RESTART_DELAY * 2, 60)  # 指数退避，上限60s
    elif cmd == "check":
        print(f"hermes: {'OK' if _hermes_available() else 'MISSING'}")
        print(f"profile: {_resolve_profile(AGENT_ID)}")
        print(f"version: {_get_version()}")
    elif cmd == "doc-sync":
        _sync_docs()
        print(f"文档同步完成 → {DOC_CACHE_DIR}")
    elif cmd == "health":
        _health_report()
        print("健康数据已上报")
    else:
        print(f"V4.7 Usage: {sys.argv[0]} <listen|check|doc-sync|health>")