feat: 提示词一致性校验(preflight)+同步机制(deploy_sync)

scripts/preflight_verify.py

@@ -143,7 +143,51 @@ def check_cron_timestamps():
             pass
     check("cron过期检查", stale_count == 0, f"{stale_count}个cron长时间未正常运行" if stale_count else "")
 
-# ── 6. DB死锁检测 ──
+# ── 7. 提示词一致性检查 ──
+def check_prompt_sync():
+    """检查注册的cron prompt版本与jobs.json实际运行的是否一致"""
+    try:
+        reg_path = "/home/hmo/projects/MoFin/data/prompts/registry.json"
+        cron_path = "/home/hmo/.hermes/profiles/position-analyst/cron/jobs.json"
+        if not os.path.exists(reg_path) or not os.path.exists(cron_path):
+            check("提示词一致性", False, "registry或jobs.json不存在")
+            return
+        with open(reg_path) as f:
+            reg = json.load(f)
+        with open(cron_path) as f:
+            crons = json.load(f).get("jobs", [])
+        # 建立cron prompt索引
+        cron_prompts = {}
+        for j in crons:
+            name = j.get("name", "")
+            pid = name.replace(" ","-").replace("（","-").replace("）","").replace("(","-").replace(")","")
+            if j.get("prompt"):
+                cron_prompts[pid] = {"name": name, "prompt": j["prompt"]}
+        # 逐个检查注册的版本文件是否与cron一致
+        drift = 0
+        for p in reg.get("prompts", []):
+            pid = p["id"]
+            if pid not in cron_prompts:
+                continue
+            cv = p.get("current_version", "v1")
+            # 找版本文件路径
+            content_path = ""
+            for v in p.get("versions", []):
+                if v.get("version") == cv:
+                    content_path = v.get("content_path", "")
+                    break
+            if not content_path or not os.path.exists(content_path):
+                drift += 1
+                continue
+            with open(content_path) as f:
+                registered = f.read()
+            actual = cron_prompts[pid]["prompt"]
+            if registered != actual:
+                drift += 1
+        check("提示词一致性: 注册版≈运行版", drift == 0,
+              f"{drift}个prompt不一致" if drift else "")
+    except Exception as e:
+        check("提示词一致性", False, str(e))
 def check_db_locks():
     """检查最近24小时是否有 database is locked 错误"""
     try:
@@ -167,6 +211,7 @@ if __name__ == "__main__":
     check_smoke()
     check_cron_timestamps()
     check_db_locks()
+    check_prompt_sync()
     
     # 汇总
     print()